diff --git a/src/db.test.ts b/src/db.test.ts index 28ff00d..b8dc4d4 100644 --- a/src/db.test.ts +++ b/src/db.test.ts @@ -1,3 +1,5 @@ +import fs from 'fs'; + import { describe, it, expect, beforeEach } from 'vitest'; import { @@ -27,6 +29,10 @@ import { storeMessage, updateTask, } from './db.js'; +import { + resolveTaskRuntimeIpcPath, + resolveTaskSessionsPath, +} from './group-folder.js'; beforeEach(() => { _initTestDatabase(); @@ -412,6 +418,47 @@ describe('task CRUD', () => { expect(getTaskById('task-3')).toBeUndefined(); }); + it('deletes task-scoped IPC and session directories when removing a task', () => { + const taskId = 'task-cleanup'; + const groupFolder = 'cleanup-group'; + const runtimeIpcDir = resolveTaskRuntimeIpcPath(groupFolder, taskId); + const taskSessionsDir = resolveTaskSessionsPath(groupFolder, taskId); + + fs.rmSync(runtimeIpcDir, { recursive: true, force: true }); + fs.rmSync(taskSessionsDir, { recursive: true, force: true }); + fs.mkdirSync(runtimeIpcDir, { recursive: true }); + fs.mkdirSync(taskSessionsDir, { recursive: true }); + + createTask({ + id: taskId, + group_folder: groupFolder, + chat_jid: 'group@g.us', + prompt: ` +[BACKGROUND CI WATCH] + +Watch target: +cleanup + +Task ID: +task-cleanup + +Check instructions: +Check the run. + `.trim(), + schedule_type: 'interval', + schedule_value: '60000', + context_mode: 'group', + next_run: null, + status: 'active', + created_at: '2024-01-01T00:00:00.000Z', + }); + + deleteTask(taskId); + + expect(fs.existsSync(runtimeIpcDir)).toBe(false); + expect(fs.existsSync(taskSessionsDir)).toBe(false); + }); + it('returns due tasks only for the requested agent type', () => { const dueAt = new Date(Date.now() - 1_000).toISOString(); diff --git a/src/db.ts b/src/db.ts index 4a1b232..6c7a1b6 100644 --- a/src/db.ts +++ b/src/db.ts @@ -9,8 +9,13 @@ import { SERVICE_ID, STORE_DIR, } from './config.js'; -import { isValidGroupFolder } from './group-folder.js'; +import { + isValidGroupFolder, + resolveTaskRuntimeIpcPath as resolveTaskRuntimeIpcPathFromGroup, + resolveTaskSessionsPath as resolveTaskSessionsPathFromGroup, +} from './group-folder.js'; import { logger } from './logger.js'; +import { getTaskRuntimeTaskId } from './task-watch-status.js'; import { NewMessage, AgentType, @@ -132,6 +137,7 @@ function createSchema(database: Database.Database): void { group_folder TEXT NOT NULL, chat_jid TEXT NOT NULL, agent_type TEXT, + max_duration_ms INTEGER, status_message_id TEXT, status_started_at TEXT, prompt TEXT NOT NULL, @@ -199,6 +205,12 @@ function createSchema(database: Database.Database): void { /* column already exists */ } + try { + database.exec(`ALTER TABLE scheduled_tasks ADD COLUMN max_duration_ms INTEGER`); + } catch { + /* column already exists */ + } + try { database.exec( `ALTER TABLE scheduled_tasks ADD COLUMN status_message_id TEXT`, @@ -863,24 +875,27 @@ export function createTask( | 'last_run' | 'last_result' | 'agent_type' + | 'max_duration_ms' | 'status_message_id' | 'status_started_at' > & { agent_type?: AgentType | null; + max_duration_ms?: number | null; status_message_id?: string | null; status_started_at?: string | null; }, ): void { db.prepare( ` - INSERT INTO scheduled_tasks (id, group_folder, chat_jid, agent_type, status_message_id, status_started_at, prompt, schedule_type, schedule_value, context_mode, next_run, status, created_at) - VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) + INSERT INTO scheduled_tasks (id, group_folder, chat_jid, agent_type, max_duration_ms, status_message_id, status_started_at, prompt, schedule_type, schedule_value, context_mode, next_run, status, created_at) + VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) `, ).run( task.id, task.group_folder, task.chat_jid, task.agent_type || SERVICE_AGENT_TYPE, + task.max_duration_ms ?? null, task.status_message_id || null, task.status_started_at || null, task.prompt, @@ -1009,9 +1024,40 @@ export function updateTaskStatusTracking( } export function deleteTask(id: string): void { + const task = getTaskById(id); // Delete child records first (FK constraint) db.prepare('DELETE FROM task_run_logs WHERE task_id = ?').run(id); db.prepare('DELETE FROM scheduled_tasks WHERE id = ?').run(id); + + if (!task) return; + + const runtimeTaskId = getTaskRuntimeTaskId(task); + if (!runtimeTaskId) return; + + const cleanupTargets = []; + try { + cleanupTargets.push( + resolveTaskRuntimeIpcPathFromGroup(task.group_folder, runtimeTaskId), + resolveTaskSessionsPathFromGroup(task.group_folder, runtimeTaskId), + ); + } catch (err) { + logger.warn( + { taskId: id, groupFolder: task.group_folder, err }, + 'Failed to resolve task-scoped cleanup paths', + ); + return; + } + + for (const cleanupPath of cleanupTargets) { + try { + fs.rmSync(cleanupPath, { recursive: true, force: true }); + } catch (err) { + logger.warn( + { taskId: id, cleanupPath, err }, + 'Failed to remove task-scoped runtime artifacts', + ); + } + } } export function getDueTasks( diff --git a/src/ipc-auth.test.ts b/src/ipc-auth.test.ts index 059c2fc..3434593 100644 --- a/src/ipc-auth.test.ts +++ b/src/ipc-auth.test.ts @@ -9,6 +9,7 @@ import { setRegisteredGroup, } from './db.js'; import { processTaskIpc, IpcDeps } from './ipc.js'; +import { DEFAULT_WATCH_CI_MAX_DURATION_MS } from './task-watch-status.js'; import { RegisteredGroup } from './types.js'; // Set up registered groups used across tests @@ -580,12 +581,32 @@ Check the run. const tasks = getAllTasks(); expect(tasks).toHaveLength(1); expect(tasks[0].schedule_type).toBe('interval'); + expect(tasks[0].max_duration_ms).toBe(DEFAULT_WATCH_CI_MAX_DURATION_MS); const nextRun = new Date(tasks[0].next_run!).getTime(); expect(nextRun).toBeGreaterThanOrEqual(before - 1000); expect(nextRun).toBeLessThanOrEqual(Date.now() + 1000); expect(deps.nudgeScheduler).toHaveBeenCalledTimes(1); }); + it('does not assign a max duration to regular scheduled tasks', async () => { + await processTaskIpc( + { + type: 'schedule_task', + prompt: 'regular interval task', + schedule_type: 'interval', + schedule_value: '60000', + targetJid: 'other@g.us', + }, + 'whatsapp_main', + true, + deps, + ); + + const tasks = getAllTasks(); + expect(tasks).toHaveLength(1); + expect(tasks[0].max_duration_ms).toBeNull(); + }); + it('rejects invalid interval (non-numeric)', async () => { await processTaskIpc( { diff --git a/src/ipc.ts b/src/ipc.ts index 8454ca3..4f4ba41 100644 --- a/src/ipc.ts +++ b/src/ipc.ts @@ -14,7 +14,10 @@ import { AvailableGroup } from './agent-runner.js'; import { createTask, deleteTask, getTaskById, updateTask } from './db.js'; import { isValidGroupFolder } from './group-folder.js'; import { logger } from './logger.js'; -import { isWatchCiTask } from './task-watch-status.js'; +import { + DEFAULT_WATCH_CI_MAX_DURATION_MS, + isWatchCiTask, +} from './task-watch-status.js'; import { RegisteredGroup } from './types.js'; export interface IpcDeps { @@ -313,6 +316,9 @@ export async function processTaskIpc( group_folder: targetFolder, chat_jid: resolvedTargetJid, agent_type: targetGroupEntry.agentType || SERVICE_AGENT_TYPE, + max_duration_ms: isWatchCiTask({ prompt: data.prompt }) + ? DEFAULT_WATCH_CI_MAX_DURATION_MS + : null, prompt: data.prompt, schedule_type: scheduleType, schedule_value: data.schedule_value, diff --git a/src/task-scheduler.test.ts b/src/task-scheduler.test.ts index 79659fc..130a869 100644 --- a/src/task-scheduler.test.ts +++ b/src/task-scheduler.test.ts @@ -838,6 +838,58 @@ Check the run. ); }); + it('deletes active tasks that exceed max duration before they run', async () => { + const enqueueTask = vi.fn(); + createTask({ + id: 'task-watch-expired', + group_folder: 'shared-group', + chat_jid: 'shared@g.us', + agent_type: 'codex', + max_duration_ms: 60_000, + prompt: ` +[BACKGROUND CI WATCH] + +Watch target: +GitHub Actions run 999999 + +Task ID: +task-watch-expired + +Check instructions: +Check the run. + `.trim(), + schedule_type: 'interval', + schedule_value: '60000', + context_mode: 'group', + next_run: new Date(Date.now() + 60_000).toISOString(), + status: 'active', + created_at: '2026-02-22T00:00:00.000Z', + }); + + startSchedulerLoop({ + serviceAgentType: 'codex', + registeredGroups: () => ({ + 'shared@g.us': { + name: 'Shared', + folder: 'shared-group', + trigger: '@Codex', + added_at: '2026-02-22T00:00:00.000Z', + agentType: 'codex', + }, + }), + getSessions: () => ({ 'shared-group': 'session-123' }), + queue: { enqueueTask } as any, + onProcess: () => {}, + sendMessage: async () => {}, + }); + + await vi.advanceTimersByTimeAsync(10); + + expect(getTaskById('task-watch-expired')).toBeUndefined(); + expect(enqueueTask).not.toHaveBeenCalled(); + expect(runAgentProcessMock).not.toHaveBeenCalled(); + }); + it('isolates both IPC and session state for isolated tasks', async () => { const dueAt = new Date(Date.now() - 60_000).toISOString(); createTask({ diff --git a/src/task-scheduler.ts b/src/task-scheduler.ts index 902a7a6..872df38 100644 --- a/src/task-scheduler.ts +++ b/src/task-scheduler.ts @@ -18,6 +18,7 @@ import { } from './agent-runner.js'; import { getAllTasks, + deleteTask, getDueTasks, getTaskById, logTaskRun, @@ -124,6 +125,31 @@ 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; @@ -795,7 +821,35 @@ export function startSchedulerLoop(deps: SchedulerDependencies): void { schedulerTickInFlight = true; try { - const dueTasks = getDueTasks(deps.serviceAgentType || SERVICE_AGENT_TYPE); + const agentType = deps.serviceAgentType || SERVICE_AGENT_TYPE; + const nowMs = Date.now(); + const activeTasks = getAllTasks(agentType).filter( + (task) => task.status === 'active', + ); + + for (const task of activeTasks) { + const currentTask = getTaskById(task.id); + if (!currentTask || currentTask.status !== 'active') { + continue; + } + + if (!hasTaskExceededMaxDuration(currentTask, nowMs)) { + continue; + } + + deleteTask(currentTask.id); + logger.warn( + { + taskId: currentTask.id, + groupFolder: currentTask.group_folder, + maxDurationMs: currentTask.max_duration_ms, + createdAt: currentTask.created_at, + }, + 'Deleted task that exceeded max duration', + ); + } + + const dueTasks = getDueTasks(agentType); if (dueTasks.length > 0) { logger.info({ count: dueTasks.length }, 'Found due tasks'); } diff --git a/src/task-watch-status.ts b/src/task-watch-status.ts index 1e9daea..edac995 100644 --- a/src/task-watch-status.ts +++ b/src/task-watch-status.ts @@ -10,6 +10,7 @@ export type WatcherStatusPhase = export const WATCH_CI_PREFIX = '[BACKGROUND CI WATCH]'; export const TASK_STATUS_MESSAGE_PREFIX = '\u2063\u2063\u2063'; +export const DEFAULT_WATCH_CI_MAX_DURATION_MS = 24 * 60 * 60 * 1000; export function isWatchCiTask(task: Pick): boolean { return task.prompt.startsWith(WATCH_CI_PREFIX); diff --git a/src/types.ts b/src/types.ts index d16be88..9f37e0e 100644 --- a/src/types.ts +++ b/src/types.ts @@ -72,6 +72,7 @@ export interface ScheduledTask { group_folder: string; chat_jid: string; agent_type: AgentType | null; + max_duration_ms?: number | null; status_message_id: string | null; status_started_at: string | null; prompt: string;