From 90baf7847fca03263ad468a6fea0ac6b40d2d216 Mon Sep 17 00:00:00 2001 From: ejclaw Date: Wed, 8 Apr 2026 01:06:47 +0900 Subject: [PATCH] Continue runtime and scheduler cleanup --- src/index.test.ts | 11 ++ src/index.ts | 8 +- src/message-runtime-dispatch.ts | 20 +- src/message-runtime-gating.test.ts | 88 +++++++++ src/message-runtime-handoffs.ts | 20 +- src/message-runtime-loop.ts | 21 +- src/message-runtime-queue.ts | 29 +-- src/message-runtime-shared.test.ts | 56 ++++++ src/message-runtime-types.ts | 29 +++ src/task-scheduler-github.ts | 155 +++++++++++++++ src/task-scheduler-runtime.ts | 127 ++++++++++++ src/task-scheduler-types.ts | 37 ++++ src/task-scheduler.ts | 308 ++--------------------------- 13 files changed, 525 insertions(+), 384 deletions(-) create mode 100644 src/message-runtime-gating.test.ts create mode 100644 src/message-runtime-shared.test.ts create mode 100644 src/message-runtime-types.ts create mode 100644 src/task-scheduler-github.ts create mode 100644 src/task-scheduler-runtime.ts create mode 100644 src/task-scheduler-types.ts diff --git a/src/index.test.ts b/src/index.test.ts index ef77dff..f09ff58 100644 --- a/src/index.test.ts +++ b/src/index.test.ts @@ -2,6 +2,7 @@ import { describe, expect, it, vi } from 'vitest'; import { editFormattedTrackedChannelMessage, + isTerminalStatusMessage, sendFormattedChannelMessage, sendFormattedTrackedChannelMessage, } from './index.js'; @@ -21,6 +22,16 @@ function makeChannel(overrides: Partial = {}): Channel { } describe('index scheduler messaging helpers', () => { + it('uses the paired verdict parser to detect terminal status messages', () => { + expect(isTerminalStatusMessage('DONE\nfinished')).toBe(true); + expect(isTerminalStatusMessage('**DONE_WITH_CONCERNS**\nneed follow-up')).toBe( + true, + ); + expect(isTerminalStatusMessage('Approved.\nlooks good')).toBe(true); + expect(isTerminalStatusMessage('LGTM\nship it')).toBe(true); + expect(isTerminalStatusMessage('random prose only')).toBe(false); + }); + it('sends formatted tracked messages through sendAndTrack', async () => { const sendAndTrack = vi.fn(async () => 'msg-123'); const channel = makeChannel({ sendAndTrack }); diff --git a/src/index.ts b/src/index.ts index 7f1c169..32db1a8 100644 --- a/src/index.ts +++ b/src/index.ts @@ -75,12 +75,10 @@ import { hasAvailableClaudeToken, initTokenRotation, } from './token-rotation.js'; +import { parseVisibleVerdict } from './paired-execution-context-shared.js'; -function isTerminalStatusMessage(text: string): boolean { - const trimmed = text.trimStart(); - return /^(?:\*\*)?(DONE(?:_WITH_CONCERNS)?|BLOCKED|NEEDS_CONTEXT)\b/.test( - trimmed, - ); +export function isTerminalStatusMessage(text: string): boolean { + return parseVisibleVerdict(text) !== 'continue'; } import { startTokenRefreshLoop, diff --git a/src/message-runtime-dispatch.ts b/src/message-runtime-dispatch.ts index 1659d96..a4dea46 100644 --- a/src/message-runtime-dispatch.ts +++ b/src/message-runtime-dispatch.ts @@ -20,9 +20,9 @@ import { } from './session-commands.js'; import { isSessionCommandSenderAllowed } from './config.js'; import { hasReviewerLease } from './service-routing.js'; +import type { ExecuteTurnFn } from './message-runtime-types.js'; import type { ScheduledPairedFollowUpIntentKind } from './paired-follow-up-scheduler.js'; import type { - AgentType, Channel, NewMessage, PairedTask, @@ -32,24 +32,6 @@ import type { type RuntimeLog = Pick; -type ExecuteTurnFn = (args: { - group: RegisteredGroup; - prompt: string; - chatJid: string; - runId: string; - channel: Channel; - startSeq: number | null; - endSeq: number | null; - deliveryRole?: PairedRoomRole; - hasHumanMessage?: boolean; - forcedRole?: PairedRoomRole; - forcedAgentType?: AgentType; -}) => Promise<{ - outputStatus: 'success' | 'error'; - deliverySucceeded: boolean; - visiblePhase: unknown; -}>; - export function enqueueGenericFollowUpAfterDeliveryRetry(args: { chatJid: string; runId: string; diff --git a/src/message-runtime-gating.test.ts b/src/message-runtime-gating.test.ts new file mode 100644 index 0000000..4a0d29c --- /dev/null +++ b/src/message-runtime-gating.test.ts @@ -0,0 +1,88 @@ +import { beforeEach, describe, expect, it, vi } from 'vitest'; +import type { RegisteredGroup } from './types.js'; + +const { + handleSessionCommandMock, + hasAllowedTriggerMock, + loggerInfoMock, +} = vi.hoisted(() => ({ + handleSessionCommandMock: vi.fn(), + hasAllowedTriggerMock: vi.fn(), + loggerInfoMock: vi.fn(), +})); + +vi.mock('./session-commands.js', () => ({ + handleSessionCommand: handleSessionCommandMock, +})); + +vi.mock('./message-runtime-rules.js', () => ({ + hasAllowedTrigger: hasAllowedTriggerMock, +})); + +vi.mock('./logger.js', () => ({ + logger: { + info: loggerInfoMock, + }, +})); + +import { handleQueuedRunGates } from './message-runtime-gating.js'; + +describe('message-runtime-gating', () => { + const baseArgs = { + chatJid: 'room-1', + group: { + folder: 'room-folder', + name: 'room', + isMain: false, + trigger: '코덱스', + added_at: new Date().toISOString(), + } satisfies RegisteredGroup, + runId: 'run-1', + missedMessages: [], + triggerPattern: /^코덱스/, + timezone: 'Asia/Seoul', + hasImplicitContinuationWindow: () => false, + sessionCommandDeps: {} as never, + }; + + beforeEach(() => { + handleSessionCommandMock.mockReset(); + hasAllowedTriggerMock.mockReset(); + loggerInfoMock.mockReset(); + }); + + it('returns session-command results directly when a command is handled', async () => { + handleSessionCommandMock.mockResolvedValue({ + handled: true, + success: false, + }); + + const result = await handleQueuedRunGates(baseArgs); + + expect(result).toEqual({ handled: true, success: false }); + expect(hasAllowedTriggerMock).not.toHaveBeenCalled(); + }); + + it('treats missing triggers as a handled no-op', async () => { + handleSessionCommandMock.mockResolvedValue({ handled: false }); + hasAllowedTriggerMock.mockReturnValue(false); + + const result = await handleQueuedRunGates(baseArgs); + + expect(result).toEqual({ handled: true, success: true }); + expect(loggerInfoMock).toHaveBeenCalledWith( + { chatJid: 'room-1', group: 'room', runId: 'run-1' }, + 'Skipping queued run because no allowed trigger was found', + ); + }); + + it('falls through when no session command is handled and the trigger is allowed', async () => { + handleSessionCommandMock.mockResolvedValue({ handled: false }); + hasAllowedTriggerMock.mockReturnValue(true); + + const result = await handleQueuedRunGates(baseArgs); + + expect(result).toEqual({ handled: false }); + expect(loggerInfoMock).not.toHaveBeenCalled(); + }); +}); diff --git a/src/message-runtime-handoffs.ts b/src/message-runtime-handoffs.ts index 3f6535f..53a51e0 100644 --- a/src/message-runtime-handoffs.ts +++ b/src/message-runtime-handoffs.ts @@ -14,30 +14,12 @@ import { resolveHandoffCursorKey, resolveHandoffRoleOverride, } from './message-runtime-shared.js'; +import type { ExecuteTurnFn } from './message-runtime-types.js'; import type { - AgentType, Channel, - PairedRoomRole, RegisteredGroup, } from './types.js'; -type ExecuteTurnFn = (args: { - group: RegisteredGroup; - prompt: string; - chatJid: string; - runId: string; - channel: Channel; - startSeq: number | null; - endSeq: number | null; - deliveryRole?: PairedRoomRole; - forcedRole?: PairedRoomRole; - forcedAgentType?: AgentType; -}) => Promise<{ - outputStatus: 'success' | 'error'; - deliverySucceeded: boolean; - visiblePhase: unknown; -}>; - export function enqueuePendingHandoffs(args: { enqueueTask: ( chatJid: string, diff --git a/src/message-runtime-loop.ts b/src/message-runtime-loop.ts index 2c9aa04..56b77c5 100644 --- a/src/message-runtime-loop.ts +++ b/src/message-runtime-loop.ts @@ -10,35 +10,16 @@ import { advanceLastAgentCursor, getProcessableMessages, } from './message-runtime-rules.js'; +import type { ExecuteTurnFn } from './message-runtime-types.js'; import type { ScheduledPairedFollowUpIntentKind } from './paired-follow-up-scheduler.js'; import { findChannel, formatMessages } from './router.js'; import type { - AgentType, Channel, NewMessage, - PairedRoomRole, PairedTask, RegisteredGroup, } from './types.js'; -type ExecuteTurnFn = (args: { - group: RegisteredGroup; - prompt: string; - chatJid: string; - runId: string; - channel: Channel; - startSeq: number | null; - endSeq: number | null; - deliveryRole?: PairedRoomRole; - hasHumanMessage?: boolean; - forcedRole?: PairedRoomRole; - forcedAgentType?: AgentType; -}) => Promise<{ - outputStatus: 'success' | 'error'; - deliverySucceeded: boolean; - visiblePhase: unknown; -}>; - export async function processMessageLoopTick(args: { assistantName: string; failureFinalText: string; diff --git a/src/message-runtime-queue.ts b/src/message-runtime-queue.ts index 96be6bb..c7855ae 100644 --- a/src/message-runtime-queue.ts +++ b/src/message-runtime-queue.ts @@ -16,37 +16,16 @@ import { resolveQueuedTurnRole, } from './message-runtime-rules.js'; import type { - AgentType, + ExecuteTurnFn, + RoleToChannelMap, +} from './message-runtime-types.js'; +import type { Channel, NewMessage, - PairedRoomRole, PairedTask, RegisteredGroup, } from './types.js'; -type ExecuteTurnFn = (args: { - group: RegisteredGroup; - prompt: string; - chatJid: string; - runId: string; - channel: Channel; - startSeq: number | null; - endSeq: number | null; - deliveryRole?: PairedRoomRole; - hasHumanMessage?: boolean; - forcedRole?: PairedRoomRole; - forcedAgentType?: AgentType; -}) => Promise<{ - outputStatus: 'success' | 'error'; - deliverySucceeded: boolean; - visiblePhase: unknown; -}>; - -type RoleToChannelMap = Record< - 'owner' | 'reviewer' | 'arbiter', - Channel | null ->; - export async function runPendingPairedTurnIfNeeded(args: { chatJid: string; group: RegisteredGroup; diff --git a/src/message-runtime-shared.test.ts b/src/message-runtime-shared.test.ts new file mode 100644 index 0000000..ce3eab5 --- /dev/null +++ b/src/message-runtime-shared.test.ts @@ -0,0 +1,56 @@ +import { describe, expect, it } from 'vitest'; + +import { + getFixedRoleChannelName, + getMissingRoleChannelMessage, + resolveHandoffCursorKey, + resolveHandoffRoleOverride, +} from './message-runtime-shared.js'; + +describe('message-runtime-shared', () => { + it('prefers explicit target role over inferred handoff metadata', () => { + expect( + resolveHandoffRoleOverride({ + target_role: 'arbiter', + intended_role: 'reviewer', + reason: 'reviewer-follow-up', + }), + ).toBe('arbiter'); + }); + + it('falls back from intended role to reason prefix when needed', () => { + expect( + resolveHandoffRoleOverride({ + target_role: null, + intended_role: 'reviewer', + reason: 'arbiter-follow-up', + }), + ).toBe('reviewer'); + expect( + resolveHandoffRoleOverride({ + target_role: null, + intended_role: null, + reason: 'arbiter-follow-up', + }), + ).toBe('arbiter'); + }); + + it('builds owner and role-scoped cursor keys', () => { + expect(resolveHandoffCursorKey('room-1')).toBe('room-1'); + expect(resolveHandoffCursorKey('room-1', 'owner')).toBe('room-1'); + expect(resolveHandoffCursorKey('room-1', 'reviewer')).toBe( + 'room-1:reviewer', + ); + }); + + it('returns fixed role channel names and user-facing missing-channel errors', () => { + expect(getFixedRoleChannelName('reviewer')).toBe('discord-review'); + expect(getFixedRoleChannelName('arbiter')).toBe('discord-arbiter'); + expect(getMissingRoleChannelMessage('reviewer')).toContain( + 'discord-review', + ); + expect(getMissingRoleChannelMessage('arbiter')).toContain( + 'discord-arbiter', + ); + }); +}); diff --git a/src/message-runtime-types.ts b/src/message-runtime-types.ts new file mode 100644 index 0000000..dcd04b7 --- /dev/null +++ b/src/message-runtime-types.ts @@ -0,0 +1,29 @@ +import type { + AgentType, + Channel, + PairedRoomRole, + RegisteredGroup, +} from './types.js'; + +export type ExecuteTurnFn = (args: { + group: RegisteredGroup; + prompt: string; + chatJid: string; + runId: string; + channel: Channel; + startSeq: number | null; + endSeq: number | null; + deliveryRole?: PairedRoomRole; + hasHumanMessage?: boolean; + forcedRole?: PairedRoomRole; + forcedAgentType?: AgentType; +}) => Promise<{ + outputStatus: 'success' | 'error'; + deliverySucceeded: boolean; + visiblePhase: unknown; +}>; + +export type RoleToChannelMap = Record< + 'owner' | 'reviewer' | 'arbiter', + Channel | null +>; diff --git a/src/task-scheduler-github.ts b/src/task-scheduler-github.ts new file mode 100644 index 0000000..32fbcc8 --- /dev/null +++ b/src/task-scheduler-github.ts @@ -0,0 +1,155 @@ +import { getErrorMessage } from './utils.js'; + +import { + deleteTask, + getTaskById, + logTaskRun, + updateTask, + updateTaskAfterRun, +} from './db.js'; +import { logger } from './logger.js'; +import { createTaskStatusTracker } from './task-status-tracker.js'; +import { extractWatchCiTarget } from './task-watch-status.js'; +import { + checkGitHubActionsRun, + computeGitHubWatcherDelayMs, + MAX_GITHUB_CONSECUTIVE_ERRORS, + parseGitHubCiMetadata, + serializeGitHubCiMetadata, +} from './github-ci.js'; +import { sendScheduledMessage } from './task-scheduler-runtime.js'; +import type { SchedulerDependencies } from './task-scheduler-types.js'; +import type { ScheduledTask } from './types.js'; + +export async function runGithubCiTask( + task: ScheduledTask, + deps: SchedulerDependencies, +): Promise { + const startTime = Date.now(); + const runAtIso = new Date().toISOString(); + let result: string | null = null; + let error: string | null = null; + let completedAndDeleted = false; + let paused = false; + const statusTracker = createTaskStatusTracker(task, { + sendTrackedMessage: deps.sendTrackedMessage, + editTrackedMessage: deps.editTrackedMessage, + }); + const parsedMetadata = parseGitHubCiMetadata(task.ci_metadata); + const metadata = parsedMetadata + ? { + ...parsedMetadata, + poll_count: (parsedMetadata.poll_count ?? 0) + 1, + last_checked_at: runAtIso, + } + : null; + + try { + await statusTracker.update('checking'); + + const check = await checkGitHubActionsRun(task); + result = check.resultSummary; + + if (metadata) { + metadata.consecutive_errors = 0; + } + + if (check.terminal) { + await statusTracker.update('completed'); + if (check.completionMessage) { + await sendScheduledMessage(deps, task.chat_jid, check.completionMessage); + } + deleteTask(task.id); + completedAndDeleted = true; + logger.info( + { + taskId: task.id, + groupFolder: task.group_folder, + durationMs: Date.now() - startTime, + }, + 'GitHub CI watcher completed and deleted', + ); + } else { + logger.info( + { + taskId: task.id, + groupFolder: task.group_folder, + result, + }, + 'GitHub CI watcher checked non-terminal run', + ); + } + } catch (err) { + error = getErrorMessage(err); + if (metadata) { + metadata.consecutive_errors = (metadata.consecutive_errors ?? 0) + 1; + } + logger.error({ taskId: task.id, error }, 'GitHub CI watcher failed'); + } + + const durationMs = Date.now() - startTime; + const currentTask = getTaskById(task.id); + const nextRun = currentTask + ? new Date( + Date.now() + computeGitHubWatcherDelayMs(currentTask, Date.now()), + ).toISOString() + : null; + + if (!currentTask) { + if (!completedAndDeleted) { + await statusTracker.update('completed'); + } + logger.debug( + { taskId: task.id }, + 'GitHub CI watcher deleted during execution, skipping persistence', + ); + return; + } + + if (metadata) { + updateTask(task.id, { ci_metadata: serializeGitHubCiMetadata(metadata) }); + } + + if ( + error && + metadata && + (metadata.consecutive_errors ?? 0) >= MAX_GITHUB_CONSECUTIVE_ERRORS + ) { + paused = true; + updateTask(task.id, { status: 'paused' }); + await deps.sendMessage( + task.chat_jid, + [ + `CI 감시 일시정지: ${extractWatchCiTarget(task.prompt) || task.id}`, + `- 사유: gh api 연속 ${metadata.consecutive_errors}회 실패`, + `- 마지막 오류: ${error.slice(0, 200)}`, + `- 태스크 ID: \`${task.id}\``, + ].join('\n'), + ); + } + + if (error && !paused) { + await statusTracker.update('retrying', nextRun); + } else if (paused) { + // Paused tasks keep their current status message state; the pause notice is sent separately. + } else if (nextRun) { + await statusTracker.update('waiting', nextRun); + } else { + await statusTracker.update('completed'); + } + + logTaskRun({ + task_id: task.id, + run_at: new Date().toISOString(), + duration_ms: durationMs, + status: error ? 'error' : 'success', + result, + error, + }); + + updateTaskAfterRun( + task.id, + nextRun, + error ? `Error: ${error}` : result || 'Completed', + ); +} diff --git a/src/task-scheduler-runtime.ts b/src/task-scheduler-runtime.ts new file mode 100644 index 0000000..1c2f864 --- /dev/null +++ b/src/task-scheduler-runtime.ts @@ -0,0 +1,127 @@ +import fs from 'fs'; + +import { writeTasksSnapshot } from './agent-runner.js'; +import { getAllTasks } from './db.js'; +import { + resolveGroupFolderPath, + resolveGroupIpcPath, + resolveTaskRuntimeIpcPath, +} from './group-folder.js'; +import { logger } from './logger.js'; +import { hasReviewerLease } from './service-routing.js'; +import { + getTaskQueueJid, + getTaskRuntimeTaskId, + shouldUseTaskScopedSession, +} from './task-watch-status.js'; +import type { AgentType, ScheduledTask } from './types.js'; +import type { + SchedulerDependencies, + TaskExecutionContext, +} from './task-scheduler-types.js'; + +export function hasTaskExceededMaxDuration( + task: Pick, + nowMs: number, +): boolean { + if ( + task.max_duration_ms === null || + task.max_duration_ms === undefined || + !Number.isFinite(task.max_duration_ms) || + task.max_duration_ms <= 0 + ) { + return false; + } + + const createdAtMs = new Date(task.created_at).getTime(); + if (!Number.isFinite(createdAtMs)) { + logger.warn( + { taskId: task.id, createdAt: task.created_at }, + 'Task has invalid created_at for max duration enforcement', + ); + return false; + } + + return nowMs - createdAtMs >= task.max_duration_ms; +} + +export async function sendScheduledMessage( + deps: SchedulerDependencies, + chatJid: string, + text: string, +): Promise { + if (!hasReviewerLease(chatJid)) { + await deps.sendMessage(chatJid, text); + return; + } + + if (!deps.sendMessageViaReviewerBot) { + throw new Error( + 'Paired-room scheduled output requires a configured reviewer Discord bot', + ); + } + + await deps.sendMessageViaReviewerBot(chatJid, text); +} + +export function resolveTaskExecutionContext( + task: ScheduledTask, + deps: SchedulerDependencies, +): TaskExecutionContext { + const groupDir = resolveGroupFolderPath(task.group_folder); + fs.mkdirSync(groupDir, { recursive: true }); + + const groups = deps.registeredGroups(); + const group = Object.values(groups).find( + (registeredGroup) => registeredGroup.folder === task.group_folder, + ); + if (!group) { + throw new Error(`Group not found: ${task.group_folder}`); + } + + const isMain = group.isMain === true; + const taskAgentType = + task.agent_type || deps.serviceAgentType || 'claude-code'; + const sessions = deps.getSessions(); + const runtimeTaskId = getTaskRuntimeTaskId(task); + const useTaskScopedSession = shouldUseTaskScopedSession(task); + const runtimeIpcDir = runtimeTaskId + ? resolveTaskRuntimeIpcPath(task.group_folder, runtimeTaskId) + : resolveGroupIpcPath(task.group_folder); + + return { + group, + groupDir, + isMain, + queueJid: getTaskQueueJid(task), + runtimeIpcDir, + runtimeTaskId, + sessionId: + task.context_mode === 'group' ? sessions[task.group_folder] : undefined, + useTaskScopedSession, + taskAgentType, + }; +} + +export function writeTaskSnapshotForGroup( + taskAgentType: AgentType, + groupFolder: string, + isMain: boolean, + runtimeTaskId?: string, +): void { + const tasks = getAllTasks(taskAgentType); + writeTasksSnapshot( + groupFolder, + 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, + })), + runtimeTaskId, + ); +} diff --git a/src/task-scheduler-types.ts b/src/task-scheduler-types.ts new file mode 100644 index 0000000..51a3065 --- /dev/null +++ b/src/task-scheduler-types.ts @@ -0,0 +1,37 @@ +import type { ChildProcess } from 'child_process'; + +import type { GroupQueue } from './group-queue.js'; +import type { AgentType, RegisteredGroup } from './types.js'; + +export interface SchedulerDependencies { + serviceAgentType?: AgentType; + registeredGroups: () => Record; + getSessions: () => Record; + queue: GroupQueue; + onProcess: ( + groupJid: string, + proc: ChildProcess, + processName: string, + ipcDir: string, + ) => void; + sendMessage: (jid: string, text: string) => Promise; + sendMessageViaReviewerBot?: (jid: string, text: string) => Promise; + sendTrackedMessage?: (jid: string, text: string) => Promise; + editTrackedMessage?: ( + jid: string, + messageId: string, + text: string, + ) => Promise; +} + +export interface TaskExecutionContext { + group: RegisteredGroup; + groupDir: string; + isMain: boolean; + queueJid: string; + runtimeIpcDir: string; + runtimeTaskId?: string; + sessionId?: string; + useTaskScopedSession: boolean; + taskAgentType: AgentType; +} diff --git a/src/task-scheduler.ts b/src/task-scheduler.ts index 14a2134..84e21f1 100644 --- a/src/task-scheduler.ts +++ b/src/task-scheduler.ts @@ -1,6 +1,4 @@ -import { ChildProcess } from 'child_process'; import { CronExpressionParser } from 'cron-parser'; -import fs from 'fs'; import { getAgentOutputText } from './agent-output.js'; import { createEvaluatedOutputHandler } from './agent-attempt.js'; import { @@ -14,7 +12,6 @@ import { ASSISTANT_NAME, SCHEDULER_POLL_INTERVAL, TIMEZONE } from './config.js'; import { AgentOutput, runAgentProcess, - writeTasksSnapshot, } from './agent-runner.js'; import { getAllTasks, @@ -25,12 +22,6 @@ import { updateTask, updateTaskAfterRun, } from './db.js'; -import { GroupQueue } from './group-queue.js'; -import { - resolveGroupFolderPath, - resolveGroupIpcPath, - resolveTaskRuntimeIpcPath, -} from './group-folder.js'; import { createScopedLogger, logger } from './logger.js'; import { createTaskStatusTracker } from './task-status-tracker.js'; import { @@ -57,19 +48,20 @@ import { import { extractWatchCiTarget, getTaskQueueJid, - getTaskRuntimeTaskId, isGitHubCiTask, - shouldUseTaskScopedSession, } from './task-watch-status.js'; -import { AgentType, RegisteredGroup, ScheduledTask } from './types.js'; -import { hasReviewerLease } from './service-routing.js'; +import { ScheduledTask } from './types.js'; import { - checkGitHubActionsRun, - computeGitHubWatcherDelayMs, - MAX_GITHUB_CONSECUTIVE_ERRORS, - parseGitHubCiMetadata, - serializeGitHubCiMetadata, -} from './github-ci.js'; + hasTaskExceededMaxDuration, + resolveTaskExecutionContext, + sendScheduledMessage, + writeTaskSnapshotForGroup, +} from './task-scheduler-runtime.js'; +import { runGithubCiTask } from './task-scheduler-github.js'; +import type { + SchedulerDependencies, + TaskExecutionContext, +} from './task-scheduler-types.js'; export { extractWatchCiTarget, getTaskQueueJid, @@ -79,6 +71,7 @@ export { renderWatchCiStatusMessage, shouldUseTaskScopedSession, } from './task-watch-status.js'; +export type { SchedulerDependencies } from './task-scheduler-types.js'; /** * Compute the next run time for a recurring task, anchored to the @@ -121,146 +114,6 @@ export function computeNextRun(task: ScheduledTask): string | null { return null; } -function hasTaskExceededMaxDuration( - task: Pick, - nowMs: number, -): boolean { - if ( - task.max_duration_ms === null || - task.max_duration_ms === undefined || - !Number.isFinite(task.max_duration_ms) || - task.max_duration_ms <= 0 - ) { - return false; - } - - const createdAtMs = new Date(task.created_at).getTime(); - if (!Number.isFinite(createdAtMs)) { - logger.warn( - { taskId: task.id, createdAt: task.created_at }, - 'Task has invalid created_at for max duration enforcement', - ); - return false; - } - - return nowMs - createdAtMs >= task.max_duration_ms; -} - -export interface SchedulerDependencies { - serviceAgentType?: AgentType; - registeredGroups: () => Record; - getSessions: () => Record; - queue: GroupQueue; - onProcess: ( - groupJid: string, - proc: ChildProcess, - processName: string, - ipcDir: string, - ) => void; - sendMessage: (jid: string, text: string) => Promise; - /** Send a message via the reviewer bot identity so the owner treats it as a peer request. */ - sendMessageViaReviewerBot?: (jid: string, text: string) => Promise; - sendTrackedMessage?: (jid: string, text: string) => Promise; - editTrackedMessage?: ( - jid: string, - messageId: string, - text: string, - ) => Promise; -} - -interface TaskExecutionContext { - group: RegisteredGroup; - groupDir: string; - isMain: boolean; - queueJid: string; - runtimeIpcDir: string; - runtimeTaskId?: string; - sessionId?: string; - useTaskScopedSession: boolean; - taskAgentType: AgentType; -} - -async function sendScheduledMessage( - deps: SchedulerDependencies, - chatJid: string, - text: string, -): Promise { - if (!hasReviewerLease(chatJid)) { - await deps.sendMessage(chatJid, text); - return; - } - - if (!deps.sendMessageViaReviewerBot) { - throw new Error( - 'Paired-room scheduled output requires a configured reviewer Discord bot', - ); - } - - await deps.sendMessageViaReviewerBot(chatJid, text); -} - -function resolveTaskExecutionContext( - task: ScheduledTask, - deps: SchedulerDependencies, -): TaskExecutionContext { - const groupDir = resolveGroupFolderPath(task.group_folder); - fs.mkdirSync(groupDir, { recursive: true }); - - const groups = deps.registeredGroups(); - const group = Object.values(groups).find( - (registeredGroup) => registeredGroup.folder === task.group_folder, - ); - if (!group) { - throw new Error(`Group not found: ${task.group_folder}`); - } - - const isMain = group.isMain === true; - const taskAgentType = - task.agent_type || deps.serviceAgentType || 'claude-code'; - const sessions = deps.getSessions(); - const runtimeTaskId = getTaskRuntimeTaskId(task); - const useTaskScopedSession = shouldUseTaskScopedSession(task); - const runtimeIpcDir = runtimeTaskId - ? resolveTaskRuntimeIpcPath(task.group_folder, runtimeTaskId) - : resolveGroupIpcPath(task.group_folder); - - return { - group, - groupDir, - isMain, - queueJid: getTaskQueueJid(task), - runtimeIpcDir, - runtimeTaskId, - sessionId: - task.context_mode === 'group' ? sessions[task.group_folder] : undefined, - useTaskScopedSession, - taskAgentType, - }; -} - -function writeTaskSnapshotForGroup( - taskAgentType: AgentType, - groupFolder: string, - isMain: boolean, - runtimeTaskId?: string, -): void { - const tasks = getAllTasks(taskAgentType); - writeTasksSnapshot( - groupFolder, - 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, - })), - runtimeTaskId, - ); -} - async function runTask( task: ScheduledTask, deps: SchedulerDependencies, @@ -629,143 +482,6 @@ async function runTask( updateTaskAfterRun(task.id, nextRun, resultSummary); } -async function runGithubCiTask( - task: ScheduledTask, - deps: SchedulerDependencies, -): Promise { - const startTime = Date.now(); - const runAtIso = new Date().toISOString(); - let result: string | null = null; - let error: string | null = null; - let completedAndDeleted = false; - let paused = false; - const statusTracker = createTaskStatusTracker(task, { - sendTrackedMessage: deps.sendTrackedMessage, - editTrackedMessage: deps.editTrackedMessage, - }); - const parsedMetadata = parseGitHubCiMetadata(task.ci_metadata); - const metadata = parsedMetadata - ? { - ...parsedMetadata, - poll_count: (parsedMetadata.poll_count ?? 0) + 1, - last_checked_at: runAtIso, - } - : null; - - try { - await statusTracker.update('checking'); - - const check = await checkGitHubActionsRun(task); - result = check.resultSummary; - - if (metadata) { - metadata.consecutive_errors = 0; - } - - if (check.terminal) { - await statusTracker.update('completed'); - if (check.completionMessage) { - await sendScheduledMessage( - deps, - task.chat_jid, - check.completionMessage, - ); - } - deleteTask(task.id); - completedAndDeleted = true; - logger.info( - { - taskId: task.id, - groupFolder: task.group_folder, - durationMs: Date.now() - startTime, - }, - 'GitHub CI watcher completed and deleted', - ); - } else { - logger.info( - { - taskId: task.id, - groupFolder: task.group_folder, - result, - }, - 'GitHub CI watcher checked non-terminal run', - ); - } - } catch (err) { - error = getErrorMessage(err); - if (metadata) { - metadata.consecutive_errors = (metadata.consecutive_errors ?? 0) + 1; - } - logger.error({ taskId: task.id, error }, 'GitHub CI watcher failed'); - } - - const durationMs = Date.now() - startTime; - const currentTask = getTaskById(task.id); - const nextRun = currentTask - ? new Date( - Date.now() + computeGitHubWatcherDelayMs(currentTask, Date.now()), - ).toISOString() - : null; - - if (!currentTask) { - if (!completedAndDeleted) { - await statusTracker.update('completed'); - } - logger.debug( - { taskId: task.id }, - 'GitHub CI watcher deleted during execution, skipping persistence', - ); - return; - } - - if (metadata) { - updateTask(task.id, { ci_metadata: serializeGitHubCiMetadata(metadata) }); - } - - if ( - error && - metadata && - (metadata.consecutive_errors ?? 0) >= MAX_GITHUB_CONSECUTIVE_ERRORS - ) { - paused = true; - updateTask(task.id, { status: 'paused' }); - await deps.sendMessage( - task.chat_jid, - [ - `CI 감시 일시정지: ${extractWatchCiTarget(task.prompt) || task.id}`, - `- 사유: gh api 연속 ${metadata.consecutive_errors}회 실패`, - `- 마지막 오류: ${error.slice(0, 200)}`, - `- 태스크 ID: \`${task.id}\``, - ].join('\n'), - ); - } - - if (error && !paused) { - await statusTracker.update('retrying', nextRun); - } else if (paused) { - // Paused tasks keep their current status message state; the pause notice is sent separately. - } else if (nextRun) { - await statusTracker.update('waiting', nextRun); - } else { - await statusTracker.update('completed'); - } - - logTaskRun({ - task_id: task.id, - run_at: new Date().toISOString(), - duration_ms: durationMs, - status: error ? 'error' : 'success', - result, - error, - }); - - updateTaskAfterRun( - task.id, - nextRun, - error ? `Error: ${error}` : result || 'Completed', - ); -} - /** * Execute one scheduler tick without timer/in-flight coordination. * This keeps the scheduler loop focused on timing while tests and debugging