mirror of
https://github.com/Ark0N/Codeman.git
synced 2026-10-05 23:19:43 +02:00
refactor(cron): rename scheduler feature to cron
Rename the recurring-jobs feature scheduler->cron to disambiguate from the legacy ScheduledRun system (/api/scheduled), which is left untouched: - ScheduledJob->CronJob, SchedulerService->CronService - /api/scheduler/jobs -> /api/cron/jobs; SSE scheduler:* -> cron:* - state keys cronJobs/cronJobRuns - files moved to src/cron/, cron-routes.ts, cron-port.ts, types/cron.ts - frontend cron-ui.js, #cronModal, menu "Cron" - docs moved to docs/cron-discovery.md + docs/cron-build-brief.md, README guides - new tests: cron-service.test.ts, cron-time.test.ts Green: tsc, lint, frontend-syntax, format, 30 cron + 9 legacy scheduled-runs tests. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01PmvZR12aX2v8K7YhqxPUAU
This commit is contained in:
Kris
@@ -0,0 +1,30 @@
|
||||
/**
|
||||
* @fileoverview Input shape for creating/updating a cron job. This is the
|
||||
* user-settable subset of `CronJob` (server-maintained bookkeeping fields
|
||||
* such as nextRunAt / lastStatus are excluded). Produced by the zod schema.
|
||||
*/
|
||||
|
||||
import type { ConcurrencyPolicy, InputMode, PromptMode, ScheduleType } from '../types/cron.js';
|
||||
import type { SessionMode } from '../types/session.js';
|
||||
|
||||
export type { CronJob, CronJobRun, CronJobRunStatus, TriggerType } from '../types/cron.js';
|
||||
|
||||
export interface CronJobInput {
|
||||
name: string;
|
||||
agentType: SessionMode;
|
||||
workingDir: string;
|
||||
launchCommand?: string;
|
||||
promptMode: PromptMode;
|
||||
promptText?: string;
|
||||
promptFilePath?: string;
|
||||
inputMode: InputMode;
|
||||
scheduleType: ScheduleType;
|
||||
runAt?: number;
|
||||
intervalMinutes?: number;
|
||||
dailyTime?: string;
|
||||
weeklyDays?: number[];
|
||||
weeklyTime?: string;
|
||||
enabled: boolean;
|
||||
notes?: string;
|
||||
concurrencyPolicy: ConcurrencyPolicy;
|
||||
}
|
||||
@@ -0,0 +1,357 @@
|
||||
/**
|
||||
* @fileoverview Cron service: CRUD for cron jobs, manual Run Now,
|
||||
* the background due-job tick, and run-history recording.
|
||||
*
|
||||
* It does NOT own session/tmux logic — it reuses Codeman's existing session
|
||||
* layer (create → addSession → setupSessionListeners → startInteractive/Shell →
|
||||
* send prompt via writeViaMux/write), mirroring the "quick start" route flow.
|
||||
*/
|
||||
|
||||
import { v4 as uuidv4 } from 'uuid';
|
||||
import { readFile } from 'node:fs/promises';
|
||||
import { statSync } from 'node:fs';
|
||||
import { Session } from '../session.js';
|
||||
import { SseEvent } from '../web/sse-events.js';
|
||||
import { getErrorMessage } from '../types/api.js';
|
||||
import { MAX_CONCURRENT_SESSIONS } from '../config/map-limits.js';
|
||||
import { CRON_READY_MAX_ATTEMPTS, CRON_READY_SETTLE_MS } from '../config/server-timing.js';
|
||||
import { computeNextRunAt, dueKeyFor } from './cron-time.js';
|
||||
import type { SessionPort, EventPort, ConfigPort, InfraPort } from '../web/ports/index.js';
|
||||
import type { CronJob, CronJobRun, CronJobRunStatus, TriggerType } from '../types/cron.js';
|
||||
import type { CronJobInput } from './cron-input.js';
|
||||
|
||||
/** The subset of the route context the cron depends on. */
|
||||
export type CronDeps = SessionPort & EventPort & ConfigPort & InfraPort;
|
||||
|
||||
const delay = (ms: number): Promise<void> => new Promise((r) => setTimeout(r, ms));
|
||||
|
||||
export class CronService {
|
||||
constructor(private readonly deps: CronDeps) {}
|
||||
|
||||
private get store() {
|
||||
return this.deps.store;
|
||||
}
|
||||
|
||||
// ───────────────────────────── Reads ─────────────────────────────
|
||||
|
||||
listJobs(): CronJob[] {
|
||||
return Object.values(this.store.getCronJobs());
|
||||
}
|
||||
|
||||
getJob(id: string): CronJob | null {
|
||||
return this.store.getCronJob(id);
|
||||
}
|
||||
|
||||
listRuns(jobId?: string): CronJobRun[] {
|
||||
const all = Object.values(this.store.getCronJobRuns());
|
||||
const filtered = jobId ? all.filter((r) => r.cronJobId === jobId) : all;
|
||||
return filtered.sort((a, b) => b.startedAt - a.startedAt);
|
||||
}
|
||||
|
||||
/** Number of active sessions of a given agent type (for the multi-session warning). */
|
||||
countActiveAgents(agentType: string): number {
|
||||
let n = 0;
|
||||
for (const s of this.deps.sessions.values()) if (s.mode === agentType) n++;
|
||||
return n;
|
||||
}
|
||||
|
||||
// ──────────────────────────── Mutations ───────────────────────────
|
||||
|
||||
createJob(input: CronJobInput): CronJob {
|
||||
const now = Date.now();
|
||||
const job: CronJob = {
|
||||
id: uuidv4(),
|
||||
name: input.name,
|
||||
agentType: input.agentType,
|
||||
workingDir: input.workingDir,
|
||||
launchCommand: input.launchCommand,
|
||||
promptMode: input.promptMode,
|
||||
promptText: input.promptText,
|
||||
promptFilePath: input.promptFilePath,
|
||||
inputMode: input.inputMode,
|
||||
scheduleType: input.scheduleType,
|
||||
runAt: input.runAt,
|
||||
intervalMinutes: input.intervalMinutes,
|
||||
dailyTime: input.dailyTime,
|
||||
weeklyDays: input.weeklyDays,
|
||||
weeklyTime: input.weeklyTime,
|
||||
enabled: input.enabled,
|
||||
notes: input.notes,
|
||||
concurrencyPolicy: input.concurrencyPolicy,
|
||||
createdAt: now,
|
||||
updatedAt: now,
|
||||
lastRunAt: null,
|
||||
nextRunAt: null,
|
||||
lastStatus: null,
|
||||
lastDueKey: null,
|
||||
};
|
||||
job.nextRunAt = job.enabled ? computeNextRunAt(job, now) : null;
|
||||
this.store.setCronJob(job.id, job);
|
||||
this.broadcastListChanged();
|
||||
return job;
|
||||
}
|
||||
|
||||
updateJob(id: string, patch: Partial<CronJobInput>): CronJob | null {
|
||||
const existing = this.getJob(id);
|
||||
if (!existing) return null;
|
||||
const now = Date.now();
|
||||
const updated: CronJob = {
|
||||
...existing,
|
||||
...patch,
|
||||
id: existing.id,
|
||||
createdAt: existing.createdAt,
|
||||
updatedAt: now,
|
||||
// Editing a job re-arms it: clear the one-time completion + dup-guard so a
|
||||
// changed schedule can fire again.
|
||||
completedOnce: false,
|
||||
lastDueKey: null,
|
||||
};
|
||||
updated.nextRunAt = updated.enabled ? computeNextRunAt(updated, now) : null;
|
||||
this.store.setCronJob(updated.id, updated);
|
||||
this.broadcastListChanged();
|
||||
return updated;
|
||||
}
|
||||
|
||||
setEnabled(id: string, enabled: boolean): CronJob | null {
|
||||
const existing = this.getJob(id);
|
||||
if (!existing) return null;
|
||||
const now = Date.now();
|
||||
existing.enabled = enabled;
|
||||
existing.updatedAt = now;
|
||||
existing.nextRunAt = enabled ? computeNextRunAt(existing, now) : null;
|
||||
this.store.setCronJob(existing.id, existing);
|
||||
this.broadcastListChanged();
|
||||
return existing;
|
||||
}
|
||||
|
||||
deleteJob(id: string): boolean {
|
||||
if (!this.getJob(id)) return false;
|
||||
this.store.removeCronJob(id);
|
||||
for (const run of this.listRuns(id)) this.store.removeCronJobRun(run.id);
|
||||
this.deps.broadcast(SseEvent.CronJobDeleted, { id });
|
||||
this.broadcastListChanged();
|
||||
return true;
|
||||
}
|
||||
|
||||
// ──────────────────────────── Execution ───────────────────────────
|
||||
|
||||
/** Manual Run Now — always launches regardless of schedule/enabled state. */
|
||||
async runNow(id: string): Promise<CronJobRun | null> {
|
||||
const job = this.getJob(id);
|
||||
if (!job) return null;
|
||||
return this.launch(job, 'manual_run_now');
|
||||
}
|
||||
|
||||
/**
|
||||
* Background tick: launch every enabled job whose next run is due. Advances
|
||||
* each job's schedule and guards against double-launching the same due time.
|
||||
*/
|
||||
async tickDueJobs(now: number = Date.now()): Promise<void> {
|
||||
for (const job of this.listJobs()) {
|
||||
if (!job.enabled || job.nextRunAt == null || job.nextRunAt > now) continue;
|
||||
|
||||
const key = dueKeyFor(job.id, job.nextRunAt);
|
||||
if (job.lastDueKey === key) {
|
||||
// This due time was already consumed (overlap/restart) — just advance.
|
||||
this.advanceAfterFire(job, now);
|
||||
continue;
|
||||
}
|
||||
|
||||
// Optional concurrency policy for AUTOMATIC runs.
|
||||
if (job.concurrencyPolicy === 'skip_if_same_agent_running' && this.countActiveAgents(job.agentType) > 0) {
|
||||
job.lastDueKey = key;
|
||||
this.advanceAfterFire(job, now);
|
||||
continue;
|
||||
}
|
||||
|
||||
job.lastDueKey = key;
|
||||
// Advance the schedule BEFORE launching so a slow launch can't be
|
||||
// re-triggered by the next tick.
|
||||
this.advanceAfterFire(job, now);
|
||||
this.launch(job, 'scheduled').catch((err) =>
|
||||
console.error(`[cron] launch failed for job ${job.id}:`, getErrorMessage(err))
|
||||
);
|
||||
}
|
||||
}
|
||||
|
||||
/** Recompute nextRunAt for loaded jobs on boot (e.g. after a restart). */
|
||||
init(): void {
|
||||
const now = Date.now();
|
||||
for (const job of this.listJobs()) {
|
||||
const isDeadOnce = job.scheduleType === 'once' && job.completedOnce;
|
||||
if (job.enabled && job.nextRunAt == null && !isDeadOnce) {
|
||||
job.nextRunAt = computeNextRunAt(job, now);
|
||||
this.store.setCronJob(job.id, job);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// ──────────────────────────── Internals ───────────────────────────
|
||||
|
||||
private advanceAfterFire(job: CronJob, now: number): void {
|
||||
if (job.scheduleType === 'once') {
|
||||
job.completedOnce = true;
|
||||
job.enabled = false;
|
||||
job.nextRunAt = null;
|
||||
} else {
|
||||
job.nextRunAt = computeNextRunAt(job, now);
|
||||
}
|
||||
job.updatedAt = now;
|
||||
this.store.setCronJob(job.id, job);
|
||||
this.broadcastListChanged();
|
||||
}
|
||||
|
||||
private async launch(job: CronJob, trigger: TriggerType): Promise<CronJobRun> {
|
||||
const run: CronJobRun = {
|
||||
id: uuidv4(),
|
||||
cronJobId: job.id,
|
||||
sessionId: null,
|
||||
sessionName: null,
|
||||
startedAt: Date.now(),
|
||||
finishedAt: null,
|
||||
status: 'created',
|
||||
triggerType: trigger,
|
||||
createdSessionUrl: null,
|
||||
};
|
||||
this.store.setCronJobRun(run.id, run);
|
||||
this.deps.broadcast(SseEvent.CronRunCreated, run);
|
||||
|
||||
// Resolve the prompt.
|
||||
let prompt: string;
|
||||
try {
|
||||
prompt = await this.resolvePrompt(job);
|
||||
} catch (err) {
|
||||
return this.failRun(job, run, `Prompt error: ${getErrorMessage(err)}`);
|
||||
}
|
||||
|
||||
// Validate working directory.
|
||||
try {
|
||||
if (!statSync(job.workingDir).isDirectory()) {
|
||||
return this.failRun(job, run, 'workingDir is not a directory');
|
||||
}
|
||||
} catch {
|
||||
return this.failRun(job, run, 'workingDir does not exist');
|
||||
}
|
||||
|
||||
// Respect the global session cap.
|
||||
if (this.deps.sessions.size >= MAX_CONCURRENT_SESSIONS) {
|
||||
return this.failRun(job, run, `Maximum concurrent sessions (${MAX_CONCURRENT_SESSIONS}) reached`);
|
||||
}
|
||||
|
||||
// Create + start the session (mirrors the quick-start route flow).
|
||||
let session: Session;
|
||||
try {
|
||||
const mode = job.agentType;
|
||||
const globalNice = await this.deps.getGlobalNiceConfig();
|
||||
const modelConfig = await this.deps.getModelConfig();
|
||||
const claudeModeConfig = await this.deps.getClaudeModeConfig();
|
||||
const model = mode !== 'shell' ? modelConfig?.defaultModel || undefined : undefined;
|
||||
session = new Session({
|
||||
workingDir: job.workingDir,
|
||||
mode,
|
||||
name: job.name,
|
||||
mux: this.deps.mux,
|
||||
useMux: true,
|
||||
niceConfig: globalNice,
|
||||
model,
|
||||
claudeMode: claudeModeConfig.claudeMode,
|
||||
allowedTools: claudeModeConfig.allowedTools,
|
||||
});
|
||||
this.deps.addSession(session);
|
||||
this.store.incrementSessionsCreated();
|
||||
this.deps.persistSessionState(session);
|
||||
await this.deps.setupSessionListeners(session);
|
||||
this.deps.broadcast(SseEvent.SessionCreated, this.deps.getSessionStateWithRespawn(session));
|
||||
if (mode === 'shell') {
|
||||
await session.startShell();
|
||||
} else {
|
||||
await session.startInteractive();
|
||||
}
|
||||
this.deps.broadcast(SseEvent.SessionInteractive, { id: session.id, mode });
|
||||
} catch (err) {
|
||||
return this.failRun(job, run, `Session launch failed: ${getErrorMessage(err)}`);
|
||||
}
|
||||
|
||||
run.sessionId = session.id;
|
||||
run.sessionName = session.name;
|
||||
run.createdSessionUrl = `/?session=${session.id}`;
|
||||
run.status = 'session_started';
|
||||
this.store.setCronJobRun(run.id, run);
|
||||
this.deps.broadcast(SseEvent.CronRunUpdated, run);
|
||||
this.updateJobLastStatus(job.id, 'session_started');
|
||||
|
||||
// Send the prompt once the CLI is ready (async; does not block the caller).
|
||||
this.sendPromptWhenReady(session.id, prompt, job, run);
|
||||
return run;
|
||||
}
|
||||
|
||||
private async resolvePrompt(job: CronJob): Promise<string> {
|
||||
if (job.promptMode === 'prompt_file_path') {
|
||||
if (!job.promptFilePath) throw new Error('prompt file path is empty');
|
||||
return readFile(job.promptFilePath, 'utf-8');
|
||||
}
|
||||
return job.promptText ?? '';
|
||||
}
|
||||
|
||||
private sendPromptWhenReady(sessionId: string, prompt: string, job: CronJob, run: CronJobRun): void {
|
||||
setImmediate(() => {
|
||||
const poll = async (): Promise<void> => {
|
||||
if (job.agentType !== 'shell') {
|
||||
for (let attempt = 0; attempt < CRON_READY_MAX_ATTEMPTS; attempt++) {
|
||||
await delay(500);
|
||||
const s = this.deps.sessions.get(sessionId);
|
||||
if (!s) return; // session was removed
|
||||
const buf = s.getTerminalBuffer().slice(-2048);
|
||||
if (buf.includes('❯') || buf.includes('tokens')) break;
|
||||
}
|
||||
await delay(CRON_READY_SETTLE_MS);
|
||||
} else {
|
||||
await delay(1000);
|
||||
}
|
||||
const s = this.deps.sessions.get(sessionId);
|
||||
if (!s) return;
|
||||
try {
|
||||
const payload = prompt.endsWith('\r') ? prompt : `${prompt}\r`;
|
||||
if (job.inputMode === 'paste') {
|
||||
s.write(payload);
|
||||
} else {
|
||||
await s.writeViaMux(payload);
|
||||
}
|
||||
run.status = 'prompt_sent';
|
||||
run.finishedAt = Date.now();
|
||||
this.store.setCronJobRun(run.id, run);
|
||||
this.deps.broadcast(SseEvent.CronRunUpdated, run);
|
||||
this.updateJobLastStatus(job.id, 'prompt_sent');
|
||||
} catch (err) {
|
||||
this.failRun(job, run, `Failed to send prompt: ${getErrorMessage(err)}`);
|
||||
}
|
||||
};
|
||||
poll().catch((err) => console.error('[cron] sendPromptWhenReady error:', getErrorMessage(err)));
|
||||
});
|
||||
}
|
||||
|
||||
private failRun(job: CronJob, run: CronJobRun, message: string): CronJobRun {
|
||||
run.status = 'failed';
|
||||
run.errorMessage = message;
|
||||
run.finishedAt = Date.now();
|
||||
this.store.setCronJobRun(run.id, run);
|
||||
this.deps.broadcast(SseEvent.CronRunUpdated, run);
|
||||
this.updateJobLastStatus(job.id, 'failed');
|
||||
return run;
|
||||
}
|
||||
|
||||
private updateJobLastStatus(jobId: string, status: CronJobRunStatus): void {
|
||||
const fresh = this.store.getCronJob(jobId);
|
||||
if (!fresh) return;
|
||||
const now = Date.now();
|
||||
fresh.lastStatus = status;
|
||||
fresh.lastRunAt = now;
|
||||
fresh.updatedAt = now;
|
||||
this.store.setCronJob(fresh.id, fresh);
|
||||
this.broadcastListChanged();
|
||||
}
|
||||
|
||||
private broadcastListChanged(): void {
|
||||
this.deps.broadcast(SseEvent.CronJobsChanged, { jobs: this.listJobs() });
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,82 @@
|
||||
/**
|
||||
* @fileoverview Pure next-run-time calculations for the cron.
|
||||
*
|
||||
* All functions are pure and take an explicit `after` timestamp (epoch ms) so
|
||||
* they are deterministic and unit-testable. Times use the SERVER'S LOCAL
|
||||
* timezone for v0.1 (per the build brief) — daily/weekly wall-clock times are
|
||||
* interpreted via the host's local time.
|
||||
*/
|
||||
|
||||
import type { CronJob } from '../types/cron.js';
|
||||
|
||||
/** Parse an 'HH:MM' (24-hour) string into hours/minutes, or null if invalid. */
|
||||
export function parseHHMM(value: string | undefined): { hours: number; minutes: number } | null {
|
||||
if (!value) return null;
|
||||
const m = /^(\d{1,2}):(\d{2})$/.exec(value.trim());
|
||||
if (!m) return null;
|
||||
const hours = Number(m[1]);
|
||||
const minutes = Number(m[2]);
|
||||
if (hours < 0 || hours > 23 || minutes < 0 || minutes > 59) return null;
|
||||
return { hours, minutes };
|
||||
}
|
||||
|
||||
/**
|
||||
* Returns the epoch-ms timestamp for `hours:minutes` (local time) on the day of
|
||||
* `base`, shifted by `dayOffset` days.
|
||||
*/
|
||||
function atLocalTime(base: number, hours: number, minutes: number, dayOffset: number): number {
|
||||
const d = new Date(base);
|
||||
d.setHours(hours, minutes, 0, 0);
|
||||
d.setDate(d.getDate() + dayOffset);
|
||||
return d.getTime();
|
||||
}
|
||||
|
||||
/**
|
||||
* Compute the next fire time strictly relevant to `after`, or null if the job
|
||||
* has no future run (e.g. a completed one-time job, or invalid config).
|
||||
*
|
||||
* For `once`, returns the absolute `runAt` (even if already in the past, so a
|
||||
* missed one-time job still fires once) until it has `completedOnce`.
|
||||
*/
|
||||
export function computeNextRunAt(job: CronJob, after: number): number | null {
|
||||
switch (job.scheduleType) {
|
||||
case 'once': {
|
||||
if (job.completedOnce) return null;
|
||||
return typeof job.runAt === 'number' ? job.runAt : null;
|
||||
}
|
||||
case 'interval': {
|
||||
const minutes = job.intervalMinutes;
|
||||
if (!minutes || minutes <= 0) return null;
|
||||
return after + minutes * 60_000;
|
||||
}
|
||||
case 'daily': {
|
||||
const t = parseHHMM(job.dailyTime);
|
||||
if (!t) return null;
|
||||
let next = atLocalTime(after, t.hours, t.minutes, 0);
|
||||
if (next <= after) next = atLocalTime(after, t.hours, t.minutes, 1);
|
||||
return next;
|
||||
}
|
||||
case 'weekly': {
|
||||
const t = parseHHMM(job.weeklyTime);
|
||||
if (!t) return null;
|
||||
const days = (job.weeklyDays ?? []).filter((d) => d >= 0 && d <= 6);
|
||||
if (days.length === 0) return null;
|
||||
for (let offset = 0; offset <= 7; offset++) {
|
||||
const cand = atLocalTime(after, t.hours, t.minutes, offset);
|
||||
if (cand > after && days.includes(new Date(cand).getDay())) return cand;
|
||||
}
|
||||
return null;
|
||||
}
|
||||
default:
|
||||
return null;
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Duplicate-launch guard key: identifies a specific due time for a job. The
|
||||
* cron records the key it last consumed so an overlapping or restarted
|
||||
* loop will not launch the same due time twice.
|
||||
*/
|
||||
export function dueKeyFor(jobId: string, fireTime: number): string {
|
||||
return `${jobId}:${fireTime}`;
|
||||
}
|
||||
Reference in New Issue
Block a user