* fix(scheduled-tasks): claim schedule occurrences across server instances Two OpenChamber servers sharing project config each armed timers and both dispatched the same daily/weekly/cron/once slot (#2710). Claim the occurrence in shared config under a cross-process write lock before creating a session. Co-authored-by: Serhii Dziupin <makeittech@users.noreply.github.com> * fix(scheduled-tasks): harden occurrence claim failure and lock ownership Address PR review blockers: release running-slot bookkeeping when claim throws, avoid silently dropping an armed occurrence after a due-slack sync, verify lock-file ownership on release, and cover real on-disk lock behavior. Co-authored-by: Serhii Dziupin <makeittech@users.noreply.github.com> * fix(scheduled-tasks): always release running slot on state-write failures Wrap runTask bookkeeping in finally so claim, manual-start, and completion lock timeouts cannot stuck-run a task; drop the diskNext claim guard that suppressed later occurrences; recover unparseable locks via mtime age. Co-authored-by: Serhii Dziupin <makeittech@users.noreply.github.com> * fix(scheduled-tasks): stop re-arming past nextRunAt and clear stuck running Only schedule future nextRunAt values so once-task losers and claim-failed paths cannot spin delay-0 retries. Clear past once nextRunAt on claim, and on completion-write failure retry terminal status so manual runNow still returns the session instead of a hard 500 with lastStatus stuck running. Co-authored-by: Serhii Dziupin <makeittech@users.noreply.github.com> * fix(scheduled-tasks): release write chain on lock acquire timeout withProjectWriteLock left the in-process promise chain pending when acquireProjectFileLock timed out, wedging every later project write and stranding runTask before finally. Always release the chain; surface persistError on run; record once claim failures in task state. Co-authored-by: Serhii Dziupin <makeittech@users.noreply.github.com> --------- Co-authored-by: Serhii Dziupin <makeittech@users.noreply.github.com>
210 lines
8.3 KiB
JavaScript
210 lines
8.3 KiB
JavaScript
import fs from 'node:fs';
|
|
import path from 'node:path';
|
|
import { OpenChamberControlError } from '../openchamber-control/error.js';
|
|
import { setLoopFileEnabled } from './loops.js';
|
|
|
|
const asNonEmptyString = (value) => {
|
|
if (typeof value !== 'string') return null;
|
|
const trimmed = value.trim();
|
|
return trimmed.length > 0 ? trimmed : null;
|
|
};
|
|
|
|
export const createScheduledTaskService = (dependencies) => {
|
|
const {
|
|
readSettingsFromDiskMigrated,
|
|
sanitizeProjects,
|
|
projectConfigRuntime,
|
|
scheduledTasksRuntime,
|
|
} = dependencies;
|
|
|
|
const listProjects = async () => {
|
|
const settings = await readSettingsFromDiskMigrated();
|
|
return sanitizeProjects(settings?.projects || []);
|
|
};
|
|
|
|
const findProjectByID = async (projectID) => {
|
|
const normalized = asNonEmptyString(projectID);
|
|
if (!normalized) throw new OpenChamberControlError('projectId is required', 400);
|
|
const projects = await listProjects();
|
|
const project = projects.find((entry) => entry.id === normalized) || null;
|
|
if (!project) throw new OpenChamberControlError('Project not found', 404);
|
|
return project;
|
|
};
|
|
|
|
const resolveProjectID = async ({ projectId, directory } = {}) => {
|
|
const requestedProjectID = asNonEmptyString(projectId);
|
|
const requestedDirectory = asNonEmptyString(directory);
|
|
if (requestedProjectID && requestedDirectory) {
|
|
throw new OpenChamberControlError('Provide only one of projectId or directory', 400);
|
|
}
|
|
if (requestedProjectID) {
|
|
await findProjectByID(requestedProjectID);
|
|
return requestedProjectID;
|
|
}
|
|
if (!requestedDirectory) throw new OpenChamberControlError('projectId or directory is required', 400);
|
|
const resolvedDirectory = path.resolve(requestedDirectory);
|
|
const projects = await listProjects();
|
|
const project = projects.find((entry) => path.resolve(entry.path) === resolvedDirectory);
|
|
if (!project) throw new OpenChamberControlError(`Project not found for directory: ${resolvedDirectory}`, 404);
|
|
return project.id;
|
|
};
|
|
|
|
const list = async (projectID) => {
|
|
await findProjectByID(projectID);
|
|
return scheduledTasksRuntime.syncProject(projectID);
|
|
};
|
|
|
|
const findLoopTask = async (projectID, taskID) => {
|
|
await findProjectByID(projectID);
|
|
const normalizedTaskID = asNonEmptyString(taskID);
|
|
if (!normalizedTaskID) throw new OpenChamberControlError('taskId is required', 400);
|
|
const tasks = await scheduledTasksRuntime.syncProject(projectID);
|
|
const task = tasks.find((entry) => entry?.id === normalizedTaskID) || null;
|
|
if (!task) throw new OpenChamberControlError('Task not found', 404);
|
|
if (!task.loopFile) throw new OpenChamberControlError('Task is not managed by a loop file', 400);
|
|
if (!fs.existsSync(task.loopFile)) throw new OpenChamberControlError('Loop file not found', 404);
|
|
return task;
|
|
};
|
|
|
|
const setLoopEnabled = async (projectID, taskID, enabled) => {
|
|
if (typeof enabled !== 'boolean') {
|
|
throw new OpenChamberControlError('enabled must be a boolean', 400);
|
|
}
|
|
const task = await findLoopTask(projectID, taskID);
|
|
try {
|
|
if (!setLoopFileEnabled(task.loopFile, enabled)) {
|
|
throw new OpenChamberControlError('Loop file must be valid before changing its enabled state', 400);
|
|
}
|
|
} catch (error) {
|
|
if (error instanceof OpenChamberControlError) throw error;
|
|
const message = error instanceof Error ? error.message : 'Failed to update loop file';
|
|
throw new OpenChamberControlError(message, 500);
|
|
}
|
|
const tasks = await scheduledTasksRuntime.syncProject(projectID);
|
|
return tasks.find((entry) => entry.id === taskID) || null;
|
|
};
|
|
|
|
const removeLoopFile = async (projectID, taskID) => {
|
|
const task = await findLoopTask(projectID, taskID);
|
|
try {
|
|
fs.unlinkSync(task.loopFile);
|
|
} catch (error) {
|
|
const message = error instanceof Error ? error.message : 'Failed to delete loop file';
|
|
throw new OpenChamberControlError(message, 500);
|
|
}
|
|
return scheduledTasksRuntime.syncProject(projectID);
|
|
};
|
|
|
|
const upsert = async (projectID, taskInput) => {
|
|
await findProjectByID(projectID);
|
|
if (!taskInput || typeof taskInput !== 'object') {
|
|
throw new OpenChamberControlError('task payload is required', 400);
|
|
}
|
|
let upserted;
|
|
try {
|
|
upserted = await projectConfigRuntime.upsertScheduledTask(projectID, taskInput);
|
|
} catch (error) {
|
|
const message = error instanceof Error ? error.message : 'Failed to save scheduled task';
|
|
const invalid = message.toLowerCase().includes('required') || message.toLowerCase().includes('invalid');
|
|
throw new OpenChamberControlError(message, invalid ? 400 : 500);
|
|
}
|
|
await scheduledTasksRuntime.syncProject(projectID);
|
|
const tasks = await projectConfigRuntime.listScheduledTasks(projectID);
|
|
return {
|
|
tasks,
|
|
task: tasks.find((task) => task.id === upserted.task.id) || upserted.task,
|
|
created: upserted.created,
|
|
};
|
|
};
|
|
|
|
const remove = async (projectID, taskID) => {
|
|
await findProjectByID(projectID);
|
|
const normalizedTaskID = asNonEmptyString(taskID);
|
|
if (!normalizedTaskID) throw new OpenChamberControlError('taskId is required', 400);
|
|
const current = await projectConfigRuntime.listScheduledTasks(projectID);
|
|
const existing = current.find((task) => task.id === normalizedTaskID) || null;
|
|
if (existing?.loopFile && fs.existsSync(existing.loopFile)) {
|
|
// Loop tasks are owned by their `.agents/loops` markdown file: deleting
|
|
// the JSON row would be silently undone by the next reconcile while the
|
|
// file exists. The file itself is the removal surface. Once the file is
|
|
// gone (the task is an orphan that the next sync would remove anyway),
|
|
// deleting the row is safe and allowed.
|
|
throw new OpenChamberControlError(
|
|
'Loop task is managed by its .agents/loops markdown file; delete the file to remove the task',
|
|
400,
|
|
);
|
|
}
|
|
const result = await projectConfigRuntime.deleteScheduledTask(projectID, normalizedTaskID);
|
|
if (!result.deleted) throw new OpenChamberControlError('Task not found', 404);
|
|
await scheduledTasksRuntime.syncProject(projectID);
|
|
return projectConfigRuntime.listScheduledTasks(projectID);
|
|
};
|
|
|
|
const run = async (projectID, taskID) => {
|
|
await findProjectByID(projectID);
|
|
const normalizedTaskID = asNonEmptyString(taskID);
|
|
if (!normalizedTaskID) throw new OpenChamberControlError('taskId is required', 400);
|
|
const result = await scheduledTasksRuntime.runNow(projectID, normalizedTaskID);
|
|
if (result.running || result.queued) {
|
|
throw new OpenChamberControlError(result.error || 'Task already running', 409);
|
|
}
|
|
if (result.skipped) throw new OpenChamberControlError('Task not found or disabled', 404);
|
|
if (!result.ok) {
|
|
throw new OpenChamberControlError(result.error || 'Task run failed', 500, { task: result.task });
|
|
}
|
|
return {
|
|
task: result.task,
|
|
sessionId: result.sessionID,
|
|
...(typeof result.persistError === 'string' && result.persistError.trim()
|
|
? { persistError: result.persistError.trim() }
|
|
: {}),
|
|
};
|
|
};
|
|
|
|
const setEnabled = async (projectID, taskID, enabled) => {
|
|
const tasks = await list(projectID);
|
|
const task = tasks.find((entry) => entry?.id === taskID);
|
|
if (!task) throw new OpenChamberControlError('Task not found', 404);
|
|
const result = await upsert(projectID, { ...task, enabled });
|
|
return result.task;
|
|
};
|
|
|
|
const status = async () => {
|
|
if (typeof scheduledTasksRuntime.getStatus === 'function') {
|
|
return scheduledTasksRuntime.getStatus();
|
|
}
|
|
const projects = await listProjects();
|
|
let enabledCount = 0;
|
|
let runningCount = 0;
|
|
for (const project of projects) {
|
|
try {
|
|
const tasks = await projectConfigRuntime.listScheduledTasks(project.id);
|
|
for (const task of tasks) {
|
|
if (task?.enabled) enabledCount += 1;
|
|
if (task?.state?.lastStatus === 'running') runningCount += 1;
|
|
}
|
|
} catch {
|
|
}
|
|
}
|
|
return {
|
|
hasEnabledScheduledTasks: enabledCount > 0,
|
|
hasRunningScheduledTasks: runningCount > 0,
|
|
enabledScheduledTasksCount: enabledCount,
|
|
runningScheduledTasksCount: runningCount,
|
|
};
|
|
};
|
|
|
|
return {
|
|
listProjects,
|
|
resolveProjectID,
|
|
list,
|
|
upsert,
|
|
remove,
|
|
run,
|
|
setEnabled,
|
|
setLoopEnabled,
|
|
removeLoopFile,
|
|
status,
|
|
};
|
|
};
|