diff --git a/src/app/api/[[...route]]/route.ts b/src/app/api/[[...route]]/route.ts index 3d4d922..13072ff 100644 --- a/src/app/api/[[...route]]/route.ts +++ b/src/app/api/[[...route]]/route.ts @@ -16,6 +16,7 @@ import { GitService } from "../../../server/core/git/services/GitService"; import { ProjectRepository } from "../../../server/core/project/infrastructure/ProjectRepository"; import { ProjectController } from "../../../server/core/project/presentation/ProjectController"; import { ProjectMetaService } from "../../../server/core/project/services/ProjectMetaService"; +import { SchedulerConfigBaseDir } from "../../../server/core/scheduler/config"; import { SchedulerService } from "../../../server/core/scheduler/domain/Scheduler"; import { SchedulerController } from "../../../server/core/scheduler/presentation/SchedulerController"; import { SessionRepository } from "../../../server/core/session/infrastructure/SessionRepository"; @@ -57,6 +58,7 @@ await Effect.runPromise( Effect.provide(ClaudeCodeService.Live), Effect.provide(GitService.Live), Effect.provide(SchedulerService.Live), + Effect.provide(SchedulerConfigBaseDir.Live), ) .pipe( /** Infrastructure */ diff --git a/src/app/projects/[projectId]/sessions/[sessionId]/components/sessionSidebar/SchedulerTab.tsx b/src/app/projects/[projectId]/sessions/[sessionId]/components/sessionSidebar/SchedulerTab.tsx index ad64133..d66bcfd 100644 --- a/src/app/projects/[projectId]/sessions/[sessionId]/components/sessionSidebar/SchedulerTab.tsx +++ b/src/app/projects/[projectId]/sessions/[sessionId]/components/sessionSidebar/SchedulerTab.tsx @@ -140,15 +140,11 @@ export const SchedulerTab: FC<{ projectId: string; sessionId: string }> = ({ if (job.schedule.type === "cron") { return `Cron: ${job.schedule.expression}`; } - const hours = Math.floor(job.schedule.delayMs / 3600000); - const minutes = Math.floor((job.schedule.delayMs % 3600000) / 60000); - const timeStr = - hours > 0 - ? `${hours}h ${minutes}m` - : minutes > 0 - ? `${minutes}m` - : `${job.schedule.delayMs}ms`; - return `${job.schedule.oneTime ? "Once" : "Recurring"}: ${timeStr}`; + if (job.schedule.type === "reserved") { + const date = new Date(job.schedule.reservedExecutionTime); + return `Reserved: ${date.toLocaleString()}`; + } + return "Unknown schedule type"; }; const formatLastRun = (lastRunAt: string | null) => { diff --git a/src/components/scheduler/SchedulerJobDialog.tsx b/src/components/scheduler/SchedulerJobDialog.tsx index d044c00..03d1bd6 100644 --- a/src/components/scheduler/SchedulerJobDialog.tsx +++ b/src/components/scheduler/SchedulerJobDialog.tsx @@ -1,7 +1,7 @@ "use client"; import { Trans, useLingui } from "@lingui/react"; -import { type FC, useCallback, useEffect, useState } from "react"; +import { type FC, useEffect, useState } from "react"; import { InlineCompletion } from "@/app/projects/[projectId]/components/chatForm/InlineCompletion"; import { useMessageCompletion } from "@/app/projects/[projectId]/components/chatForm/useMessageCompletion"; import { Button } from "@/components/ui/button"; @@ -40,8 +40,6 @@ export interface SchedulerJobDialogProps { isSubmitting?: boolean; } -type DelayUnit = "minutes" | "hours" | "days"; - export const SchedulerJobDialog: FC = ({ open, onOpenChange, @@ -53,10 +51,13 @@ export const SchedulerJobDialog: FC = ({ const { _, i18n } = useLingui(); const [name, setName] = useState(""); - const [scheduleType, setScheduleType] = useState<"cron" | "fixed">("cron"); + const [scheduleType, setScheduleType] = useState<"cron" | "reserved">("cron"); const [cronExpression, setCronExpression] = useState("0 9 * * *"); - const [delayValue, setDelayValue] = useState(60); // 60 minutes default - const [delayUnit, setDelayUnit] = useState("minutes"); + const [reservedDateTime, setReservedDateTime] = useState(() => { + const now = new Date(); + now.setHours(now.getHours() + 1); + return now.toISOString().slice(0, 16); + }); const [messageContent, setMessageContent] = useState(""); const [enabled, setEnabled] = useState(true); const [concurrencyPolicy, setConcurrencyPolicy] = useState<"skip" | "run">( @@ -66,36 +67,6 @@ export const SchedulerJobDialog: FC = ({ // Message completion hook const completion = useMessageCompletion(); - // Convert delay value and unit to milliseconds - const delayToMs = useCallback((value: number, unit: DelayUnit): number => { - switch (unit) { - case "minutes": - return value * 60 * 1000; - case "hours": - return value * 60 * 60 * 1000; - case "days": - return value * 24 * 60 * 60 * 1000; - } - }, []); - - // Convert milliseconds to delay value and unit - const msToDelay = useCallback( - (ms: number): { value: number; unit: DelayUnit } => { - const minutes = ms / (60 * 1000); - const hours = ms / (60 * 60 * 1000); - const days = ms / (24 * 60 * 60 * 1000); - - if (days >= 1 && days === Math.floor(days)) { - return { value: days, unit: "days" }; - } - if (hours >= 1 && hours === Math.floor(hours)) { - return { value: hours, unit: "hours" }; - } - return { value: minutes, unit: "minutes" }; - }, - [], - ); - // Initialize form with job data when editing useEffect(() => { if (job) { @@ -103,42 +74,82 @@ export const SchedulerJobDialog: FC = ({ setScheduleType(job.schedule.type); if (job.schedule.type === "cron") { setCronExpression(job.schedule.expression); - } else { - const { value, unit } = msToDelay(job.schedule.delayMs); - setDelayValue(value); - setDelayUnit(unit); + setConcurrencyPolicy(job.schedule.concurrencyPolicy); + } else if (job.schedule.type === "reserved") { + // Convert UTC time to local time for display + const date = new Date(job.schedule.reservedExecutionTime); + const year = date.getFullYear(); + const month = String(date.getMonth() + 1).padStart(2, "0"); + const day = String(date.getDate()).padStart(2, "0"); + const hours = String(date.getHours()).padStart(2, "0"); + const minutes = String(date.getMinutes()).padStart(2, "0"); + setReservedDateTime(`${year}-${month}-${day}T${hours}:${minutes}`); } setMessageContent(job.message.content); setEnabled(job.enabled); - setConcurrencyPolicy(job.concurrencyPolicy); } else { // Reset form for new job setName(""); setScheduleType("cron"); setCronExpression("0 9 * * *"); - setDelayValue(60); - setDelayUnit("minutes"); + const now = new Date(); + now.setHours(now.getHours() + 1); + const year = now.getFullYear(); + const month = String(now.getMonth() + 1).padStart(2, "0"); + const day = String(now.getDate()).padStart(2, "0"); + const hours = String(now.getHours()).padStart(2, "0"); + const minutes = String(now.getMinutes()).padStart(2, "0"); + setReservedDateTime(`${year}-${month}-${day}T${hours}:${minutes}`); setMessageContent(""); setEnabled(true); setConcurrencyPolicy("skip"); } - }, [job, msToDelay]); + }, [job]); const handleSubmit = () => { - const delayMs = delayToMs(delayValue, delayUnit); const newJob: NewSchedulerJob = { name, schedule: scheduleType === "cron" - ? { type: "cron", expression: cronExpression } - : { type: "fixed", delayMs, oneTime: true }, + ? { + type: "cron", + expression: cronExpression, + concurrencyPolicy, + } + : { + type: "reserved", + // datetime-local returns "YYYY-MM-DDTHH:mm" in local time + // We need to treat this as local time and convert to UTC + reservedExecutionTime: (() => { + // datetime-local format: "YYYY-MM-DDTHH:mm" + // Parse as local time and convert to ISO string (UTC) + const match = reservedDateTime.match( + /^(\d{4})-(\d{2})-(\d{2})T(\d{2}):(\d{2})$/, + ); + if (!match) { + throw new Error("Invalid datetime format"); + } + const year = Number(match[1]); + const month = Number(match[2]); + const day = Number(match[3]); + const hours = Number(match[4]); + const minutes = Number(match[5]); + const localDate = new Date( + year, + month - 1, + day, + hours, + minutes, + ); + return localDate.toISOString(); + })(), + }, message: { content: messageContent, projectId, baseSessionId: null, }, enabled, - concurrencyPolicy, }; onSubmit(newJob); @@ -220,7 +231,7 @@ export const SchedulerJobDialog: FC = ({ - setDelayValue(Number.parseInt(e.target.value, 10)) - } - disabled={isSubmitting} - className="flex-1" - placeholder="60" + + setReservedDateTime(e.target.value)} + disabled={isSubmitting} + />

@@ -358,40 +340,42 @@ export const SchedulerJobDialog: FC = ({

- {/* Concurrency Policy */} -
- - -
+ {/* Concurrency Policy (only for cron schedules) */} + {scheduleType === "cron" && ( +
+ + +
+ )} diff --git a/src/server/core/scheduler/config.test.ts b/src/server/core/scheduler/config.test.ts index b483041..c83da96 100644 --- a/src/server/core/scheduler/config.test.ts +++ b/src/server/core/scheduler/config.test.ts @@ -9,17 +9,29 @@ import { getConfigPath, initializeConfig, readConfig, + SchedulerConfigBaseDir, writeConfig, } from "./config"; import type { SchedulerConfig } from "./schema"; describe("scheduler config", () => { let testDir: string; - const testLayer = Layer.mergeAll(NodeFileSystem.layer, NodePath.layer); + let testLayer: Layer.Layer< + FileSystem.FileSystem | Path.Path | SchedulerConfigBaseDir + >; beforeEach(async () => { testDir = join(tmpdir(), `scheduler-test-${Date.now()}`); await mkdir(testDir, { recursive: true }); + + // Use test directory as base for config files + const testConfigBaseDir = Layer.succeed(SchedulerConfigBaseDir, testDir); + + testLayer = Layer.mergeAll( + NodeFileSystem.layer, + NodePath.layer, + testConfigBaseDir, + ); }); afterEach(async () => { @@ -31,7 +43,8 @@ describe("scheduler config", () => { getConfigPath.pipe(Effect.provide(testLayer)), ); - expect(result).toContain(".claude-code-viewer/scheduler/config.json"); + expect(result).toContain("/scheduler/schedules.json"); + expect(result).toContain(testDir); }); test("writeConfig and readConfig work correctly", async () => { @@ -43,6 +56,7 @@ describe("scheduler config", () => { schedule: { type: "cron", expression: "0 0 * * *", + concurrencyPolicy: "skip", }, message: { content: "test message", @@ -50,7 +64,6 @@ describe("scheduler config", () => { baseSessionId: null, }, enabled: true, - concurrencyPolicy: "skip", createdAt: "2025-10-25T00:00:00Z", lastRunAt: null, lastRunStatus: null, diff --git a/src/server/core/scheduler/config.ts b/src/server/core/scheduler/config.ts index 69a3005..83e42f6 100644 --- a/src/server/core/scheduler/config.ts +++ b/src/server/core/scheduler/config.ts @@ -1,6 +1,6 @@ import { homedir } from "node:os"; import { FileSystem, Path } from "@effect/platform"; -import { Data, Effect } from "effect"; +import { Context, Data, Effect, Layer } from "effect"; import { type SchedulerConfig, schedulerConfigSchema } from "./schema"; class ConfigFileNotFoundError extends Data.TaggedError( @@ -15,11 +15,18 @@ class ConfigParseError extends Data.TaggedError("ConfigParseError")<{ }> {} const CONFIG_DIR = "scheduler"; -const CONFIG_FILE = "config.json"; +const CONFIG_FILE = "schedules.json"; + +// Service to provide base directory (for testing) +export class SchedulerConfigBaseDir extends Context.Tag( + "SchedulerConfigBaseDir", +)() { + static Live = Layer.succeed(this, `${homedir()}/.claude-code-viewer`); +} export const getConfigPath = Effect.gen(function* () { const path = yield* Path.Path; - const baseDir = path.resolve(homedir(), ".claude-code-viewer"); + const baseDir = yield* SchedulerConfigBaseDir; return path.join(baseDir, CONFIG_DIR, CONFIG_FILE); }); diff --git a/src/server/core/scheduler/domain/Job.test.ts b/src/server/core/scheduler/domain/Job.test.ts index 7bb31ce..c84d511 100644 --- a/src/server/core/scheduler/domain/Job.test.ts +++ b/src/server/core/scheduler/domain/Job.test.ts @@ -1,16 +1,19 @@ import { describe, expect, test } from "vitest"; import type { SchedulerJob } from "../schema"; -import { calculateFixedDelay, shouldExecuteJob } from "./Job"; +import { calculateReservedDelay, shouldExecuteJob } from "./Job"; describe("shouldExecuteJob", () => { test("returns false when job is disabled", () => { const job: SchedulerJob = { id: "test-job", name: "Test Job", - schedule: { type: "cron", expression: "* * * * *" }, + schedule: { + type: "cron", + expression: "* * * * *", + concurrencyPolicy: "skip", + }, message: { content: "test", projectId: "proj-1", baseSessionId: null }, enabled: false, - concurrencyPolicy: "skip", createdAt: "2025-10-25T00:00:00Z", lastRunAt: null, lastRunStatus: null, @@ -23,10 +26,13 @@ describe("shouldExecuteJob", () => { const job: SchedulerJob = { id: "test-job", name: "Test Job", - schedule: { type: "cron", expression: "* * * * *" }, + schedule: { + type: "cron", + expression: "* * * * *", + concurrencyPolicy: "skip", + }, message: { content: "test", projectId: "proj-1", baseSessionId: null }, enabled: true, - concurrencyPolicy: "skip", createdAt: "2025-10-25T00:00:00Z", lastRunAt: null, lastRunStatus: null, @@ -35,14 +41,16 @@ describe("shouldExecuteJob", () => { expect(shouldExecuteJob(job, new Date())).toBe(true); }); - test("returns false for oneTime fixed job that has already run", () => { + test("returns false for reserved job that has already run", () => { const job: SchedulerJob = { id: "test-job", name: "Test Job", - schedule: { type: "fixed", delayMs: 60000, oneTime: true }, + schedule: { + type: "reserved", + reservedExecutionTime: "2025-10-25T00:01:00Z", + }, message: { content: "test", projectId: "proj-1", baseSessionId: null }, enabled: true, - concurrencyPolicy: "skip", createdAt: "2025-10-25T00:00:00Z", lastRunAt: "2025-10-25T00:01:00Z", lastRunStatus: "success", @@ -51,18 +59,19 @@ describe("shouldExecuteJob", () => { expect(shouldExecuteJob(job, new Date())).toBe(false); }); - test("returns false for oneTime fixed job when scheduled time has not arrived", () => { - const createdAt = new Date("2025-10-25T00:00:00Z"); + test("returns false for reserved job when scheduled time has not arrived", () => { const now = new Date("2025-10-25T00:00:30Z"); const job: SchedulerJob = { id: "test-job", name: "Test Job", - schedule: { type: "fixed", delayMs: 60000, oneTime: true }, + schedule: { + type: "reserved", + reservedExecutionTime: "2025-10-25T00:01:00Z", + }, message: { content: "test", projectId: "proj-1", baseSessionId: null }, enabled: true, - concurrencyPolicy: "skip", - createdAt: createdAt.toISOString(), + createdAt: "2025-10-25T00:00:00Z", lastRunAt: null, lastRunStatus: null, }; @@ -70,98 +79,88 @@ describe("shouldExecuteJob", () => { expect(shouldExecuteJob(job, now)).toBe(false); }); - test("returns true for oneTime fixed job when scheduled time has arrived", () => { - const createdAt = new Date("2025-10-25T00:00:00Z"); + test("returns true for reserved job when scheduled time has arrived", () => { const now = new Date("2025-10-25T00:01:01Z"); const job: SchedulerJob = { id: "test-job", name: "Test Job", - schedule: { type: "fixed", delayMs: 60000, oneTime: true }, + schedule: { + type: "reserved", + reservedExecutionTime: "2025-10-25T00:01:00Z", + }, message: { content: "test", projectId: "proj-1", baseSessionId: null }, enabled: true, - concurrencyPolicy: "skip", - createdAt: createdAt.toISOString(), + createdAt: "2025-10-25T00:00:00Z", lastRunAt: null, lastRunStatus: null, }; expect(shouldExecuteJob(job, now)).toBe(true); }); - - test("returns true for recurring fixed job", () => { - const job: SchedulerJob = { - id: "test-job", - name: "Test Job", - schedule: { type: "fixed", delayMs: 60000, oneTime: false }, - message: { content: "test", projectId: "proj-1", baseSessionId: null }, - enabled: true, - concurrencyPolicy: "skip", - createdAt: "2025-10-25T00:00:00Z", - lastRunAt: null, - lastRunStatus: null, - }; - - expect(shouldExecuteJob(job, new Date())).toBe(true); - }); }); -describe("calculateFixedDelay", () => { +describe("calculateReservedDelay", () => { test("calculates delay correctly for future scheduled time", () => { - const createdAt = new Date("2025-10-25T00:00:00Z"); const now = new Date("2025-10-25T00:00:30Z"); const job: SchedulerJob = { id: "test-job", name: "Test Job", - schedule: { type: "fixed", delayMs: 60000, oneTime: true }, + schedule: { + type: "reserved", + reservedExecutionTime: "2025-10-25T00:01:00Z", + }, message: { content: "test", projectId: "proj-1", baseSessionId: null }, enabled: true, - concurrencyPolicy: "skip", - createdAt: createdAt.toISOString(), - lastRunAt: null, - lastRunStatus: null, - }; - - const delay = calculateFixedDelay(job, now); - expect(delay).toBe(30000); - }); - - test("returns 0 for past scheduled time", () => { - const createdAt = new Date("2025-10-25T00:00:00Z"); - const now = new Date("2025-10-25T00:02:00Z"); - - const job: SchedulerJob = { - id: "test-job", - name: "Test Job", - schedule: { type: "fixed", delayMs: 60000, oneTime: true }, - message: { content: "test", projectId: "proj-1", baseSessionId: null }, - enabled: true, - concurrencyPolicy: "skip", - createdAt: createdAt.toISOString(), - lastRunAt: null, - lastRunStatus: null, - }; - - const delay = calculateFixedDelay(job, now); - expect(delay).toBe(0); - }); - - test("throws error for non-fixed schedule type", () => { - const job: SchedulerJob = { - id: "test-job", - name: "Test Job", - schedule: { type: "cron", expression: "* * * * *" }, - message: { content: "test", projectId: "proj-1", baseSessionId: null }, - enabled: true, - concurrencyPolicy: "skip", createdAt: "2025-10-25T00:00:00Z", lastRunAt: null, lastRunStatus: null, }; - expect(() => calculateFixedDelay(job, new Date())).toThrow( - "Job schedule type must be fixed", + const delay = calculateReservedDelay(job, now); + expect(delay).toBe(30000); + }); + + test("returns 0 for past scheduled time", () => { + const now = new Date("2025-10-25T00:02:00Z"); + + const job: SchedulerJob = { + id: "test-job", + name: "Test Job", + schedule: { + type: "reserved", + reservedExecutionTime: "2025-10-25T00:01:00Z", + }, + message: { content: "test", projectId: "proj-1", baseSessionId: null }, + enabled: true, + createdAt: "2025-10-25T00:00:00Z", + lastRunAt: null, + lastRunStatus: null, + }; + + const delay = calculateReservedDelay(job, now); + expect(delay).toBe(0); + }); + + test("throws error for non-reserved schedule type", () => { + const job: SchedulerJob = { + id: "test-job", + name: "Test Job", + schedule: { + type: "cron", + expression: "* * * * *", + concurrencyPolicy: "skip", + }, + message: { content: "test", projectId: "proj-1", baseSessionId: null }, + enabled: true, + createdAt: "2025-10-25T00:00:00Z", + lastRunAt: null, + lastRunStatus: null, + }; + + expect(() => calculateReservedDelay(job, new Date())).toThrow( + "Job schedule type must be reserved", ); }); }); diff --git a/src/server/core/scheduler/domain/Job.ts b/src/server/core/scheduler/domain/Job.ts index 4537ae7..b985677 100644 --- a/src/server/core/scheduler/domain/Job.ts +++ b/src/server/core/scheduler/domain/Job.ts @@ -20,23 +20,15 @@ export const executeJob = (job: SchedulerJob) => ); } - if (message.baseSessionId === null) { - yield* lifeCycleService.startTask({ - baseSession: { - cwd: project.meta.projectPath, - projectId: message.projectId, - sessionId: undefined, - }, - userConfig, - message: message.content, - }); - } else { - yield* lifeCycleService.continueTask({ - sessionProcessId: message.baseSessionId, - message: message.content, - baseSessionId: message.baseSessionId, - }); - } + yield* lifeCycleService.startTask({ + baseSession: { + cwd: project.meta.projectPath, + projectId: message.projectId, + sessionId: message.baseSessionId ?? undefined, + }, + userConfig, + message: message.content, + }); }); export const shouldExecuteJob = (job: SchedulerJob, now: Date): boolean => { @@ -48,26 +40,28 @@ export const shouldExecuteJob = (job: SchedulerJob, now: Date): boolean => { return true; } - if (job.schedule.type === "fixed" && job.schedule.oneTime) { + if (job.schedule.type === "reserved") { + // Reserved jobs are one-time, skip if already executed if (job.lastRunStatus !== null) { return false; } - const createdAt = new Date(job.createdAt); - const scheduledTime = new Date(createdAt.getTime() + job.schedule.delayMs); + const scheduledTime = new Date(job.schedule.reservedExecutionTime); return now >= scheduledTime; } return true; }; -export const calculateFixedDelay = (job: SchedulerJob, now: Date): number => { - if (job.schedule.type !== "fixed") { - throw new Error("Job schedule type must be fixed"); +export const calculateReservedDelay = ( + job: SchedulerJob, + now: Date, +): number => { + if (job.schedule.type !== "reserved") { + throw new Error("Job schedule type must be reserved"); } - const createdAt = new Date(job.createdAt); - const scheduledTime = new Date(createdAt.getTime() + job.schedule.delayMs); + const scheduledTime = new Date(job.schedule.reservedExecutionTime); const delay = scheduledTime.getTime() - now.getTime(); return Math.max(0, delay); diff --git a/src/server/core/scheduler/domain/Scheduler.test.ts b/src/server/core/scheduler/domain/Scheduler.test.ts index 4059771..2c8452d 100644 --- a/src/server/core/scheduler/domain/Scheduler.test.ts +++ b/src/server/core/scheduler/domain/Scheduler.test.ts @@ -1,5 +1,5 @@ -import { mkdir, rm, unlink } from "node:fs/promises"; -import { homedir, tmpdir } from "node:os"; +import { mkdir, rm } from "node:fs/promises"; +import { tmpdir } from "node:os"; import { join } from "node:path"; import { NodeContext, NodeFileSystem, NodePath } from "@effect/platform-node"; import { Effect, Layer } from "effect"; @@ -9,6 +9,7 @@ import { ClaudeCodeSessionProcessService } from "../../claude-code/services/Clau import { EnvService } from "../../platform/services/EnvService"; import { UserConfigService } from "../../platform/services/UserConfigService"; import { ProjectRepository } from "../../project/infrastructure/ProjectRepository"; +import { SchedulerConfigBaseDir } from "../config"; import type { NewSchedulerJob } from "../schema"; import { SchedulerService } from "./Scheduler"; @@ -66,37 +67,42 @@ describe("SchedulerService", () => { getEnv: () => Effect.succeed(undefined), } as never); - const baseLayers = Layer.mergeAll( - NodeFileSystem.layer, - NodePath.layer, - NodeContext.layer, - mockSessionProcessService, - mockLifeCycleService, - mockProjectRepository, - mockUserConfigService, - mockEnvService, - ); - - const testLayer = Layer.mergeAll(SchedulerService.Live, baseLayers).pipe( - Layer.provide(baseLayers), - ); + let testConfigBaseDir: Layer.Layer; + let testLayer: Layer.Layer< + | import("@effect/platform").FileSystem.FileSystem + | import("@effect/platform").Path.Path + | import("@effect/platform-node").NodeContext.NodeContext + | ClaudeCodeSessionProcessService + | ClaudeCodeLifeCycleService + | ProjectRepository + | UserConfigService + | EnvService + | SchedulerConfigBaseDir + | SchedulerService + >; beforeEach(async () => { testDir = join(tmpdir(), `scheduler-test-${Date.now()}`); await mkdir(testDir, { recursive: true }); - // Clean up existing config file - const configPath = join( - homedir(), - ".claude-code-viewer", - "scheduler", - "config.json", + // Use test directory as base for config files + testConfigBaseDir = Layer.succeed(SchedulerConfigBaseDir, testDir); + + const baseLayers = Layer.mergeAll( + NodeFileSystem.layer, + NodePath.layer, + NodeContext.layer, + mockSessionProcessService, + mockLifeCycleService, + mockProjectRepository, + mockUserConfigService, + mockEnvService, + testConfigBaseDir, + ); + + testLayer = Layer.mergeAll(SchedulerService.Live, baseLayers).pipe( + Layer.provideMerge(baseLayers), ); - try { - await unlink(configPath); - } catch { - // Ignore if file doesn't exist - } }); afterEach(async () => { @@ -106,14 +112,17 @@ describe("SchedulerService", () => { test("addJob creates a new job with generated id", async () => { const newJob: NewSchedulerJob = { name: "Test Job", - schedule: { type: "cron", expression: "0 0 * * *" }, + schedule: { + type: "cron", + expression: "0 0 * * *", + concurrencyPolicy: "skip", + }, message: { content: "test message", projectId: "project-1", baseSessionId: null, }, enabled: false, - concurrencyPolicy: "skip", }; const result = await Effect.runPromise( @@ -134,14 +143,17 @@ describe("SchedulerService", () => { test("getJobs returns all jobs", async () => { const newJob: NewSchedulerJob = { name: "Test Job", - schedule: { type: "cron", expression: "0 0 * * *" }, + schedule: { + type: "cron", + expression: "0 0 * * *", + concurrencyPolicy: "skip", + }, message: { content: "test message", projectId: "project-1", baseSessionId: null, }, enabled: false, - concurrencyPolicy: "skip", }; const result = await Effect.runPromise( @@ -159,14 +171,17 @@ describe("SchedulerService", () => { test("updateJob modifies an existing job", async () => { const newJob: NewSchedulerJob = { name: "Test Job", - schedule: { type: "cron", expression: "0 0 * * *" }, + schedule: { + type: "cron", + expression: "0 0 * * *", + concurrencyPolicy: "skip", + }, message: { content: "test message", projectId: "project-1", baseSessionId: null, }, enabled: false, - concurrencyPolicy: "skip", }; const result = await Effect.runPromise( @@ -186,14 +201,17 @@ describe("SchedulerService", () => { test("deleteJob removes a job", async () => { const newJob: NewSchedulerJob = { name: "Test Job", - schedule: { type: "cron", expression: "0 0 * * *" }, + schedule: { + type: "cron", + expression: "0 0 * * *", + concurrencyPolicy: "skip", + }, message: { content: "test message", projectId: "project-1", baseSessionId: null, }, enabled: false, - concurrencyPolicy: "skip", }; const result = await Effect.runPromise( diff --git a/src/server/core/scheduler/domain/Scheduler.ts b/src/server/core/scheduler/domain/Scheduler.ts index bf61186..15269eb 100644 --- a/src/server/core/scheduler/domain/Scheduler.ts +++ b/src/server/core/scheduler/domain/Scheduler.ts @@ -1,4 +1,3 @@ -import { randomUUID } from "node:crypto"; import { Context, Cron, @@ -10,6 +9,7 @@ import { Ref, Schedule, } from "effect"; +import { ulid } from "ulid"; import type { InferEffect } from "../../../lib/effect/types"; import { initializeConfig, readConfig, writeConfig } from "../config"; import type { @@ -18,7 +18,7 @@ import type { SchedulerJob, UpdateSchedulerJob, } from "../schema"; -import { calculateFixedDelay, executeJob } from "./Job"; +import { calculateReservedDelay, executeJob } from "./Job"; class SchedulerJobNotFoundError extends Data.TaggedError( "SchedulerJobNotFoundError", @@ -74,42 +74,33 @@ const LayerImpl = Effect.gen(function* () { yield* Ref.update(fibersRef, (fibers) => new Map(fibers).set(job.id, fiber), ); - } else if (job.schedule.type === "fixed") { - // For oneTime jobs, skip scheduling if already executed - if (job.schedule.oneTime && job.lastRunStatus !== null) { + } else if (job.schedule.type === "reserved") { + // For reserved jobs, skip scheduling if already executed + if (job.lastRunStatus !== null) { return; } - const delay = calculateFixedDelay(job, now); + const delay = calculateReservedDelay(job, now); const delayDuration = Duration.millis(delay); - if (job.schedule.oneTime) { - const fiber = yield* Effect.delay( - runJobWithConcurrencyControl(job), - delayDuration, - ).pipe(Effect.forkDaemon); + const fiber = yield* Effect.delay( + runJobWithConcurrencyControl(job), + delayDuration, + ).pipe(Effect.forkDaemon); - yield* Ref.update(fibersRef, (fibers) => - new Map(fibers).set(job.id, fiber), - ); - } else { - const schedule = Schedule.spaced(delayDuration); - - const fiber = yield* Effect.repeat( - runJobWithConcurrencyControl(job), - schedule, - ).pipe(Effect.forkDaemon); - - yield* Ref.update(fibersRef, (fibers) => - new Map(fibers).set(job.id, fiber), - ); - } + yield* Ref.update(fibersRef, (fibers) => + new Map(fibers).set(job.id, fiber), + ); } }); const runJobWithConcurrencyControl = (job: SchedulerJob) => Effect.gen(function* () { - if (job.concurrencyPolicy === "skip") { + // Check concurrency policy (only for cron jobs) + if ( + job.schedule.type === "cron" && + job.schedule.concurrencyPolicy === "skip" + ) { const runningJobs = yield* Ref.get(runningJobsRef); if (runningJobs.has(job.id)) { return; @@ -118,6 +109,35 @@ const LayerImpl = Effect.gen(function* () { yield* Ref.update(runningJobsRef, (jobs) => new Set(jobs).add(job.id)); + // For reserved jobs, delete after execution without updating status + if (job.schedule.type === "reserved") { + const result = yield* executeJob(job).pipe( + Effect.matchEffect({ + onSuccess: () => Effect.void, + onFailure: () => Effect.void, + }), + ); + yield* Ref.update(runningJobsRef, (jobs) => { + const newJobs = new Set(jobs); + newJobs.delete(job.id); + return newJobs; + }); + + // Delete reserved job after execution (skip fiber stop, just delete from config) + yield* deleteJobFromConfig(job.id).pipe( + Effect.catchAll((error) => { + console.error( + `[Scheduler] Failed to delete reserved job ${job.id}:`, + error, + ); + return Effect.void; + }), + ); + + return result; + } + + // For non-reserved jobs, update status const result = yield* executeJob(job).pipe( Effect.matchEffect({ onSuccess: () => @@ -223,7 +243,7 @@ const LayerImpl = Effect.gen(function* () { ); const job: SchedulerJob = { ...newJob, - id: randomUUID(), + id: ulid(), createdAt: new Date().toISOString(), lastRunAt: null, lastRunStatus: null, @@ -278,6 +298,29 @@ const LayerImpl = Effect.gen(function* () { return updatedJob; }); + const deleteJobFromConfig = (jobId: string) => + Effect.gen(function* () { + const config = yield* readConfig.pipe( + Effect.catchTags({ + ConfigFileNotFoundError: () => + initializeConfig.pipe(Effect.map(() => ({ jobs: [] }))), + ConfigParseError: () => + initializeConfig.pipe(Effect.map(() => ({ jobs: [] }))), + }), + ); + const job = config.jobs.find((j) => j.id === jobId); + + if (job === undefined) { + return yield* Effect.fail(new SchedulerJobNotFoundError({ jobId })); + } + + const updatedConfig: SchedulerConfig = { + jobs: config.jobs.filter((j) => j.id !== jobId), + }; + + yield* writeConfig(updatedConfig); + }); + const deleteJob = (jobId: string) => Effect.gen(function* () { const config = yield* readConfig.pipe( @@ -295,12 +338,7 @@ const LayerImpl = Effect.gen(function* () { } yield* stopJob(jobId); - - const updatedConfig: SchedulerConfig = { - jobs: config.jobs.filter((j) => j.id !== jobId), - }; - - yield* writeConfig(updatedConfig); + yield* deleteJobFromConfig(jobId); }); return { diff --git a/src/server/core/scheduler/schema.ts b/src/server/core/scheduler/schema.ts index 7b9b5a5..11eb1d5 100644 --- a/src/server/core/scheduler/schema.ts +++ b/src/server/core/scheduler/schema.ts @@ -1,20 +1,23 @@ import { z } from "zod"; +// Concurrency policy (for cron jobs only) +export const concurrencyPolicySchema = z.enum(["skip", "run"]); + // Schedule type discriminated union export const cronScheduleSchema = z.object({ type: z.literal("cron"), expression: z.string(), + concurrencyPolicy: concurrencyPolicySchema, }); -export const fixedScheduleSchema = z.object({ - type: z.literal("fixed"), - delayMs: z.number().int().positive(), - oneTime: z.boolean(), +export const reservedScheduleSchema = z.object({ + type: z.literal("reserved"), + reservedExecutionTime: z.iso.datetime(), }); export const scheduleSchema = z.discriminatedUnion("type", [ cronScheduleSchema, - fixedScheduleSchema, + reservedScheduleSchema, ]); // Message configuration @@ -27,9 +30,6 @@ export const messageConfigSchema = z.object({ // Job status export const jobStatusSchema = z.enum(["success", "failed"]); -// Concurrency policy -export const concurrencyPolicySchema = z.enum(["skip", "run"]); - // Scheduler job export const schedulerJobSchema = z.object({ id: z.string(), @@ -37,7 +37,6 @@ export const schedulerJobSchema = z.object({ schedule: scheduleSchema, message: messageConfigSchema, enabled: z.boolean(), - concurrencyPolicy: concurrencyPolicySchema, createdAt: z.string().datetime(), lastRunAt: z.string().datetime().nullable(), lastRunStatus: jobStatusSchema.nullable(), @@ -50,7 +49,7 @@ export const schedulerConfigSchema = z.object({ // Type exports export type CronSchedule = z.infer; -export type FixedSchedule = z.infer; +export type ReservedSchedule = z.infer; export type Schedule = z.infer; export type MessageConfig = z.infer; export type JobStatus = z.infer; @@ -68,7 +67,6 @@ export const newSchedulerJobSchema = schedulerJobSchema }) .extend({ enabled: z.boolean().default(true), - concurrencyPolicy: concurrencyPolicySchema.default("skip"), }); export type NewSchedulerJob = z.infer; @@ -79,7 +77,6 @@ export const updateSchedulerJobSchema = schedulerJobSchema.partial().pick({ schedule: true, message: true, enabled: true, - concurrencyPolicy: true, }); export type UpdateSchedulerJob = z.infer; diff --git a/src/server/hono/route.ts b/src/server/hono/route.ts index 8d07fdf..b7b3229 100644 --- a/src/server/hono/route.ts +++ b/src/server/hono/route.ts @@ -19,7 +19,12 @@ import { EnvService } from "../core/platform/services/EnvService"; import { UserConfigService } from "../core/platform/services/UserConfigService"; import type { ProjectRepository } from "../core/project/infrastructure/ProjectRepository"; import { ProjectController } from "../core/project/presentation/ProjectController"; +import type { SchedulerConfigBaseDir } from "../core/scheduler/config"; import { SchedulerController } from "../core/scheduler/presentation/SchedulerController"; +import { + newSchedulerJobSchema, + updateSchedulerJobSchema, +} from "../core/scheduler/schema"; import type { VirtualConversationDatabase } from "../core/session/infrastructure/VirtualConversationDatabase"; import { SessionController } from "../core/session/presentation/SessionController"; import type { SessionMetaService } from "../core/session/services/SessionMetaService"; @@ -60,6 +65,7 @@ export const routes = (app: HonoAppType) => | UserConfigService | ClaudeCodeLifeCycleService | ProjectRepository + | SchedulerConfigBaseDir >(); if ((yield* envService.getEnv("NEXT_PHASE")) !== "phase-production-build") { @@ -460,30 +466,7 @@ export const routes = (app: HonoAppType) => .post( "/scheduler/jobs", - zValidator( - "json", - z.object({ - name: z.string(), - schedule: z.discriminatedUnion("type", [ - z.object({ - type: z.literal("cron"), - expression: z.string(), - }), - z.object({ - type: z.literal("fixed"), - delayMs: z.number().int().positive(), - oneTime: z.boolean(), - }), - ]), - message: z.object({ - content: z.string(), - projectId: z.string(), - baseSessionId: z.string().nullable(), - }), - enabled: z.boolean().default(true), - concurrencyPolicy: z.enum(["skip", "run"]).default("skip"), - }), - ), + zValidator("json", newSchedulerJobSchema), async (c) => { const response = await effectToResponse( c, @@ -499,34 +482,7 @@ export const routes = (app: HonoAppType) => .patch( "/scheduler/jobs/:id", - zValidator( - "json", - z.object({ - name: z.string().optional(), - schedule: z - .discriminatedUnion("type", [ - z.object({ - type: z.literal("cron"), - expression: z.string(), - }), - z.object({ - type: z.literal("fixed"), - delayMs: z.number().int().positive(), - oneTime: z.boolean(), - }), - ]) - .optional(), - message: z - .object({ - content: z.string(), - projectId: z.string(), - baseSessionId: z.string().nullable(), - }) - .optional(), - enabled: z.boolean().optional(), - concurrencyPolicy: z.enum(["skip", "run"]).optional(), - }), - ), + zValidator("json", updateSchedulerJobSchema), async (c) => { const response = await effectToResponse( c,