Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
164 changes: 164 additions & 0 deletions apps/server/src/scheduledTasks/ScheduledTaskService.schedule.test.ts
Original file line number Diff line number Diff line change
Expand Up @@ -392,3 +392,167 @@ it.effect(
}).pipe(Effect.provide(ScheduledTaskService.layer.pipe(Layer.provide(layerDependencies))));
}).pipe(Effect.provide(SqlitePersistence.layerMemory)),
);

it.effect("a fixed-time slot missed while the server was off shows on the task", () =>
Effect.gen(function* () {
const at = (hour: number, minute = 0, day = 1) =>
DateTime.makeZonedUnsafe(
{ year: 2026, month: 7, day, hour, minute, second: 0, millisecond: 0 },
{ timeZone: DateTime.zoneMakeLocal(), adjustForTimeZone: true },
);
yield* TestClock.setTime(DateTime.toEpochMillis(at(8)));
const launches = yield* Ref.make(0);
const layerDependencies = Layer.mergeAll(
NodeCrypto.layer,
Scheduler.layer,
ManagedPolicy.layerTest(),
Layer.mock(ThreadLaunchService.ThreadLaunchService)({
launch: () =>
Ref.update(launches, (count) => count + 1).pipe(
Effect.as({ threadId: "thread-missed-slot", resumed: false } as never),
),
}),
Layer.mock(ManagedProjectFolders.ManagedProjectFolders)({
namedProjectsRoot: "/projects",
isScratchProject: () => Effect.succeed(false),
}),
Layer.mock(ThreadManagementService.ThreadManagementService)({}),
Layer.mock(SecretRequests.SecretRequests)({}),
);
const withServer = <A, E, R>(effect: Effect.Effect<A, E, R>) =>
effect.pipe(
Effect.provide(ScheduledTaskService.layer.pipe(Layer.provide(layerDependencies))),
);

const taskId = yield* withServer(
Effect.gen(function* () {
const service = yield* ScheduledTaskService.ScheduledTaskService;
const { task } = yield* service.upsert(
yield* decodeUpsertInput({
title: "Morning bookkeeping",
prompt: "Update yesterday's notes.",
enabled: true,
schedule: { type: "fixed_time", timeOfDay: "09:00" },
projectId: "project-missed-slot",
workspaceStrategy: { type: "root" },
modelSelection: { instanceId: "codex", model: "gpt-5.4" },
runtimeMode: "full-access",
interactionMode: "default",
}),
);
yield* service.runNow({ id: task.id });
return task.id;
}),
);

// The server is off across the 09:00 slot and comes back at 09:30.
yield* TestClock.setTime(DateTime.toEpochMillis(at(9, 30)));
const [listed] = yield* withServer(
Effect.gen(function* () {
const service = yield* ScheduledTaskService.ScheduledTaskService;
yield* TestClock.adjust("6 seconds");
return (yield* service.list()).tasks;
}),
);

expect(listed?.id).toBe(taskId);
expect(yield* Ref.get(launches)).toBe(1);
expect(listed?.nextRunAt).toBe(DateTime.formatIso(DateTime.toUtc(at(9, 0, 2))));
// The skipped slot must not read as the earlier manual run's clean success.
expect({ status: listed?.lastRunStatus, error: listed?.lastRunError }).not.toEqual({
status: "succeeded",
error: null,
});
}).pipe(Effect.provide(SqlitePersistence.layerMemory)),
);

it.effect(
"a run made while the scheduler was busy is not overwritten by its stale missed slot",
() =>
Effect.gen(function* () {
const at = (hour: number, minute = 0, day = 1) =>
DateTime.makeZonedUnsafe(
{ year: 2026, month: 7, day, hour, minute, second: 0, millisecond: 0 },
{ timeZone: DateTime.zoneMakeLocal(), adjustForTimeZone: true },
);
yield* TestClock.setTime(DateTime.toEpochMillis(at(6)));
const slowStarted = yield* Deferred.make<void>();
const releaseSlow = yield* Deferred.make<void>();
const lastProcessed = yield* Deferred.make<void>();
const layerDependencies = Layer.mergeAll(
NodeCrypto.layer,
Scheduler.layer,
ManagedPolicy.layerTest(),
Layer.mock(ThreadLaunchService.ThreadLaunchService)({
launch: (input) =>
(input.title === "Slow"
? Deferred.succeed(slowStarted, undefined).pipe(
Effect.andThen(Deferred.await(releaseSlow)),
)
: input.title === "Last"
? Deferred.succeed(lastProcessed, undefined)
: Effect.void
).pipe(Effect.as({ threadId: `thread-${input.title}`, resumed: false } as never)),
}),
Layer.mock(ManagedProjectFolders.ManagedProjectFolders)({
namedProjectsRoot: "/projects",
isScratchProject: () => Effect.succeed(false),
}),
Layer.mock(ThreadManagementService.ThreadManagementService)({}),
Layer.mock(SecretRequests.SecretRequests)({}),
);
const withServer = <A, E, R>(effect: Effect.Effect<A, E, R>) =>
effect.pipe(
Effect.provide(ScheduledTaskService.layer.pipe(Layer.provide(layerDependencies))),
);
const upsert = (
title: string,
schedule: { type: "interval"; everyMs: number } | { type: "fixed_time"; timeOfDay: string },
) =>
Effect.gen(function* () {
const service = yield* ScheduledTaskService.ScheduledTaskService;
const { task } = yield* service.upsert(
yield* decodeUpsertInput({
title,
prompt: "Update the notes.",
enabled: true,
schedule,
projectId: "project-stale-slot",
workspaceStrategy: { type: "root" },
modelSelection: { instanceId: "codex", model: "gpt-5.4" },
runtimeMode: "full-access",
interactionMode: "default",
}),
);
return task.id;
});

// Due at 07:00, 08:00 and 08:30; the server is off until 09:30.
const morningId = yield* withServer(
Effect.gen(function* () {
yield* upsert("Slow", { type: "interval", everyMs: 60 * 60 * 1000 });
const id = yield* upsert("Morning", { type: "fixed_time", timeOfDay: "08:00" });
yield* upsert("Last", { type: "interval", everyMs: 150 * 60 * 1000 });
return id;
}),
);
yield* TestClock.setTime(DateTime.toEpochMillis(at(9, 30)));
const morning = yield* withServer(
Effect.gen(function* () {
const service = yield* ScheduledTaskService.ScheduledTaskService;
yield* TestClock.adjust("6 seconds");
yield* Deferred.await(slowStarted);
yield* service.runNow({ id: morningId });
yield* Deferred.succeed(releaseSlow, undefined);
yield* Deferred.await(lastProcessed);
return (yield* service.list()).tasks.find((task) => task.id === morningId);
}),
);

expect({ status: morning?.lastRunStatus, error: morning?.lastRunError }).toEqual({
status: "succeeded",
error: null,
});
expect(morning?.nextRunAt).toBe(DateTime.formatIso(DateTime.toUtc(at(8, 0, 2))));
}).pipe(Effect.provide(SqlitePersistence.layerMemory)),
);
89 changes: 89 additions & 0 deletions apps/server/src/scheduledTasks/ScheduledTaskService.test.ts
Original file line number Diff line number Diff line change
Expand Up @@ -17,6 +17,7 @@ import * as Effect from "effect/Effect";
import * as Exit from "effect/Exit";
import * as Layer from "effect/Layer";
import * as Logger from "effect/Logger";
import * as Option from "effect/Option";
import * as SqlClient from "effect/sql/SqlClient";

import * as Fiber from "effect/Fiber";
Expand Down Expand Up @@ -418,6 +419,94 @@ it.effect("thread cleanup notifies subscribers and a finishing run cannot recrea
}).pipe(Effect.provide(SqlitePersistence.layerMemory)),
);

it.effect.each([
{ status: "failed", error: "Weekly usage limit reached" },
{ status: "interrupted", error: "The run was stopped." },
{ status: "cancelled", error: "The run was cancelled." },
] as const)("a run that ends $status after its message was accepted records it", (outcome) =>
Effect.gen(function* () {
const projectId = ProjectId.make("project:test");
const ended = yield* Deferred.make<void>();
const sent = yield* Ref.make("");
const run = { id: "run:2", status: outcome.status, rootNodeId: "node:root" };
const layer = ScheduledTaskService.layer.pipe(
Layer.provide(
Layer.mergeAll(
ManagedPolicy.layerTest(),
Layer.mock(ThreadLaunchService.ThreadLaunchService)({}),
Layer.mock(ThreadManagementService.ThreadManagementService)({
sendToThread: (input) => Ref.set(sent, input.messageId).pipe(Effect.as({} as never)),
getThreadShell: () => Effect.succeed(threadShell(projectId)),
getProjectThreadRecords: () =>
Effect.map(
Ref.get(sent),
(userMessageId) =>
({
runs: [
{ id: "run:1", status: "completed", userMessageId: "message:earlier" },
{ ...run, userMessageId },
],
messages: [{ id: userMessageId, runId: "run:2" }],
turnItems: [
{
type: "error",
status: "failed",
runId: "run:2",
nodeId: "node:root",
failure: { message: "Weekly usage limit reached" },
},
],
}) as never,
),
waitForThread: (input) =>
Deferred.await(ended).pipe(
Effect.as({
threadId: input.threadId,
run: { ...run, id: input.runId, userMessageId: "" },
timedOut: false,
} as never),
),
}),
Layer.mock(SecretRequests.SecretRequests)({}),
Layer.mock(ManagedProjectFolders.ManagedProjectFolders)({
namedProjectsRoot: "/projects",
isScratchProject: () => Effect.succeed(false),
}),
NodeCrypto.layer,
Scheduler.layer,
),
),
);
yield* Effect.gen(function* () {
const service = yield* ScheduledTaskService.ScheduledTaskService;
const { task } = yield* service.upsert({
id: ScheduledTaskId.make("task:quota"),
title: "Quota bookkeeping",
prompt: "Update the notes",
enabled: true,
schedule: { type: "interval", everyMs: 60_000 },
projectId,
threadId: ThreadId.make("thread:quota"),
workspaceStrategy: { type: "root" },
modelSelection: { instanceId: ProviderInstanceId.make("codex"), model: "gpt-5.4" },
runtimeMode: "full-access",
interactionMode: "default",
});
yield* service.runNow({ id: task.id });
const failed = yield* service.subscribeList().pipe(
Stream.map(({ tasks }) => tasks.find((candidate) => candidate.id === task.id)),
Stream.filter((current) => current?.lastRunStatus === "failed"),
Stream.runHead,
Effect.forkChild,
);
yield* Deferred.succeed(ended, undefined);
const recorded = Option.getOrUndefined(yield* Fiber.join(failed));
assert.equal(recorded?.lastRunError, outcome.error);
assert.equal(recorded?.runCount, 1);
}).pipe(Effect.provide(layer));
}).pipe(Effect.provide(SqlitePersistence.layerMemory)),
);

it.effect("a bound run follows its thread when promotion moves it mid-send", () =>
Effect.gen(function* () {
const scratch = ProjectId.make("project:scratch");
Expand Down
Loading
Loading