fix: add watcher TTL and cleanup task-scoped artifacts
This commit is contained in:
@@ -1,3 +1,5 @@
|
|||||||
|
import fs from 'fs';
|
||||||
|
|
||||||
import { describe, it, expect, beforeEach } from 'vitest';
|
import { describe, it, expect, beforeEach } from 'vitest';
|
||||||
|
|
||||||
import {
|
import {
|
||||||
@@ -27,6 +29,10 @@ import {
|
|||||||
storeMessage,
|
storeMessage,
|
||||||
updateTask,
|
updateTask,
|
||||||
} from './db.js';
|
} from './db.js';
|
||||||
|
import {
|
||||||
|
resolveTaskRuntimeIpcPath,
|
||||||
|
resolveTaskSessionsPath,
|
||||||
|
} from './group-folder.js';
|
||||||
|
|
||||||
beforeEach(() => {
|
beforeEach(() => {
|
||||||
_initTestDatabase();
|
_initTestDatabase();
|
||||||
@@ -412,6 +418,47 @@ describe('task CRUD', () => {
|
|||||||
expect(getTaskById('task-3')).toBeUndefined();
|
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', () => {
|
it('returns due tasks only for the requested agent type', () => {
|
||||||
const dueAt = new Date(Date.now() - 1_000).toISOString();
|
const dueAt = new Date(Date.now() - 1_000).toISOString();
|
||||||
|
|
||||||
|
|||||||
52
src/db.ts
52
src/db.ts
@@ -9,8 +9,13 @@ import {
|
|||||||
SERVICE_ID,
|
SERVICE_ID,
|
||||||
STORE_DIR,
|
STORE_DIR,
|
||||||
} from './config.js';
|
} 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 { logger } from './logger.js';
|
||||||
|
import { getTaskRuntimeTaskId } from './task-watch-status.js';
|
||||||
import {
|
import {
|
||||||
NewMessage,
|
NewMessage,
|
||||||
AgentType,
|
AgentType,
|
||||||
@@ -132,6 +137,7 @@ function createSchema(database: Database.Database): void {
|
|||||||
group_folder TEXT NOT NULL,
|
group_folder TEXT NOT NULL,
|
||||||
chat_jid TEXT NOT NULL,
|
chat_jid TEXT NOT NULL,
|
||||||
agent_type TEXT,
|
agent_type TEXT,
|
||||||
|
max_duration_ms INTEGER,
|
||||||
status_message_id TEXT,
|
status_message_id TEXT,
|
||||||
status_started_at TEXT,
|
status_started_at TEXT,
|
||||||
prompt TEXT NOT NULL,
|
prompt TEXT NOT NULL,
|
||||||
@@ -199,6 +205,12 @@ function createSchema(database: Database.Database): void {
|
|||||||
/* column already exists */
|
/* column already exists */
|
||||||
}
|
}
|
||||||
|
|
||||||
|
try {
|
||||||
|
database.exec(`ALTER TABLE scheduled_tasks ADD COLUMN max_duration_ms INTEGER`);
|
||||||
|
} catch {
|
||||||
|
/* column already exists */
|
||||||
|
}
|
||||||
|
|
||||||
try {
|
try {
|
||||||
database.exec(
|
database.exec(
|
||||||
`ALTER TABLE scheduled_tasks ADD COLUMN status_message_id TEXT`,
|
`ALTER TABLE scheduled_tasks ADD COLUMN status_message_id TEXT`,
|
||||||
@@ -863,24 +875,27 @@ export function createTask(
|
|||||||
| 'last_run'
|
| 'last_run'
|
||||||
| 'last_result'
|
| 'last_result'
|
||||||
| 'agent_type'
|
| 'agent_type'
|
||||||
|
| 'max_duration_ms'
|
||||||
| 'status_message_id'
|
| 'status_message_id'
|
||||||
| 'status_started_at'
|
| 'status_started_at'
|
||||||
> & {
|
> & {
|
||||||
agent_type?: AgentType | null;
|
agent_type?: AgentType | null;
|
||||||
|
max_duration_ms?: number | null;
|
||||||
status_message_id?: string | null;
|
status_message_id?: string | null;
|
||||||
status_started_at?: string | null;
|
status_started_at?: string | null;
|
||||||
},
|
},
|
||||||
): void {
|
): void {
|
||||||
db.prepare(
|
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)
|
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 (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
|
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
|
||||||
`,
|
`,
|
||||||
).run(
|
).run(
|
||||||
task.id,
|
task.id,
|
||||||
task.group_folder,
|
task.group_folder,
|
||||||
task.chat_jid,
|
task.chat_jid,
|
||||||
task.agent_type || SERVICE_AGENT_TYPE,
|
task.agent_type || SERVICE_AGENT_TYPE,
|
||||||
|
task.max_duration_ms ?? null,
|
||||||
task.status_message_id || null,
|
task.status_message_id || null,
|
||||||
task.status_started_at || null,
|
task.status_started_at || null,
|
||||||
task.prompt,
|
task.prompt,
|
||||||
@@ -1009,9 +1024,40 @@ export function updateTaskStatusTracking(
|
|||||||
}
|
}
|
||||||
|
|
||||||
export function deleteTask(id: string): void {
|
export function deleteTask(id: string): void {
|
||||||
|
const task = getTaskById(id);
|
||||||
// Delete child records first (FK constraint)
|
// Delete child records first (FK constraint)
|
||||||
db.prepare('DELETE FROM task_run_logs WHERE task_id = ?').run(id);
|
db.prepare('DELETE FROM task_run_logs WHERE task_id = ?').run(id);
|
||||||
db.prepare('DELETE FROM scheduled_tasks WHERE 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(
|
export function getDueTasks(
|
||||||
|
|||||||
@@ -9,6 +9,7 @@ import {
|
|||||||
setRegisteredGroup,
|
setRegisteredGroup,
|
||||||
} from './db.js';
|
} from './db.js';
|
||||||
import { processTaskIpc, IpcDeps } from './ipc.js';
|
import { processTaskIpc, IpcDeps } from './ipc.js';
|
||||||
|
import { DEFAULT_WATCH_CI_MAX_DURATION_MS } from './task-watch-status.js';
|
||||||
import { RegisteredGroup } from './types.js';
|
import { RegisteredGroup } from './types.js';
|
||||||
|
|
||||||
// Set up registered groups used across tests
|
// Set up registered groups used across tests
|
||||||
@@ -580,12 +581,32 @@ Check the run.
|
|||||||
const tasks = getAllTasks();
|
const tasks = getAllTasks();
|
||||||
expect(tasks).toHaveLength(1);
|
expect(tasks).toHaveLength(1);
|
||||||
expect(tasks[0].schedule_type).toBe('interval');
|
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();
|
const nextRun = new Date(tasks[0].next_run!).getTime();
|
||||||
expect(nextRun).toBeGreaterThanOrEqual(before - 1000);
|
expect(nextRun).toBeGreaterThanOrEqual(before - 1000);
|
||||||
expect(nextRun).toBeLessThanOrEqual(Date.now() + 1000);
|
expect(nextRun).toBeLessThanOrEqual(Date.now() + 1000);
|
||||||
expect(deps.nudgeScheduler).toHaveBeenCalledTimes(1);
|
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 () => {
|
it('rejects invalid interval (non-numeric)', async () => {
|
||||||
await processTaskIpc(
|
await processTaskIpc(
|
||||||
{
|
{
|
||||||
|
|||||||
@@ -14,7 +14,10 @@ import { AvailableGroup } from './agent-runner.js';
|
|||||||
import { createTask, deleteTask, getTaskById, updateTask } from './db.js';
|
import { createTask, deleteTask, getTaskById, updateTask } from './db.js';
|
||||||
import { isValidGroupFolder } from './group-folder.js';
|
import { isValidGroupFolder } from './group-folder.js';
|
||||||
import { logger } from './logger.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';
|
import { RegisteredGroup } from './types.js';
|
||||||
|
|
||||||
export interface IpcDeps {
|
export interface IpcDeps {
|
||||||
@@ -313,6 +316,9 @@ export async function processTaskIpc(
|
|||||||
group_folder: targetFolder,
|
group_folder: targetFolder,
|
||||||
chat_jid: resolvedTargetJid,
|
chat_jid: resolvedTargetJid,
|
||||||
agent_type: targetGroupEntry.agentType || SERVICE_AGENT_TYPE,
|
agent_type: targetGroupEntry.agentType || SERVICE_AGENT_TYPE,
|
||||||
|
max_duration_ms: isWatchCiTask({ prompt: data.prompt })
|
||||||
|
? DEFAULT_WATCH_CI_MAX_DURATION_MS
|
||||||
|
: null,
|
||||||
prompt: data.prompt,
|
prompt: data.prompt,
|
||||||
schedule_type: scheduleType,
|
schedule_type: scheduleType,
|
||||||
schedule_value: data.schedule_value,
|
schedule_value: data.schedule_value,
|
||||||
|
|||||||
@@ -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 () => {
|
it('isolates both IPC and session state for isolated tasks', async () => {
|
||||||
const dueAt = new Date(Date.now() - 60_000).toISOString();
|
const dueAt = new Date(Date.now() - 60_000).toISOString();
|
||||||
createTask({
|
createTask({
|
||||||
|
|||||||
@@ -18,6 +18,7 @@ import {
|
|||||||
} from './agent-runner.js';
|
} from './agent-runner.js';
|
||||||
import {
|
import {
|
||||||
getAllTasks,
|
getAllTasks,
|
||||||
|
deleteTask,
|
||||||
getDueTasks,
|
getDueTasks,
|
||||||
getTaskById,
|
getTaskById,
|
||||||
logTaskRun,
|
logTaskRun,
|
||||||
@@ -124,6 +125,31 @@ export function computeNextRun(task: ScheduledTask): string | null {
|
|||||||
return null;
|
return null;
|
||||||
}
|
}
|
||||||
|
|
||||||
|
function hasTaskExceededMaxDuration(
|
||||||
|
task: Pick<ScheduledTask, 'id' | 'created_at' | 'max_duration_ms'>,
|
||||||
|
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 {
|
export interface SchedulerDependencies {
|
||||||
serviceAgentType?: AgentType;
|
serviceAgentType?: AgentType;
|
||||||
registeredGroups: () => Record<string, RegisteredGroup>;
|
registeredGroups: () => Record<string, RegisteredGroup>;
|
||||||
@@ -795,7 +821,35 @@ export function startSchedulerLoop(deps: SchedulerDependencies): void {
|
|||||||
schedulerTickInFlight = true;
|
schedulerTickInFlight = true;
|
||||||
|
|
||||||
try {
|
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) {
|
if (dueTasks.length > 0) {
|
||||||
logger.info({ count: dueTasks.length }, 'Found due tasks');
|
logger.info({ count: dueTasks.length }, 'Found due tasks');
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -10,6 +10,7 @@ export type WatcherStatusPhase =
|
|||||||
|
|
||||||
export const WATCH_CI_PREFIX = '[BACKGROUND CI WATCH]';
|
export const WATCH_CI_PREFIX = '[BACKGROUND CI WATCH]';
|
||||||
export const TASK_STATUS_MESSAGE_PREFIX = '\u2063\u2063\u2063';
|
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<ScheduledTask, 'prompt'>): boolean {
|
export function isWatchCiTask(task: Pick<ScheduledTask, 'prompt'>): boolean {
|
||||||
return task.prompt.startsWith(WATCH_CI_PREFIX);
|
return task.prompt.startsWith(WATCH_CI_PREFIX);
|
||||||
|
|||||||
@@ -72,6 +72,7 @@ export interface ScheduledTask {
|
|||||||
group_folder: string;
|
group_folder: string;
|
||||||
chat_jid: string;
|
chat_jid: string;
|
||||||
agent_type: AgentType | null;
|
agent_type: AgentType | null;
|
||||||
|
max_duration_ms?: number | null;
|
||||||
status_message_id: string | null;
|
status_message_id: string | null;
|
||||||
status_started_at: string | null;
|
status_started_at: string | null;
|
||||||
prompt: string;
|
prompt: string;
|
||||||
|
|||||||
Reference in New Issue
Block a user