diff --git a/apps/server/src/scheduledTasks/ScheduledTaskService.schedule.test.ts b/apps/server/src/scheduledTasks/ScheduledTaskService.schedule.test.ts index a6adcdaba9108..537d071853e34 100644 --- a/apps/server/src/scheduledTasks/ScheduledTaskService.schedule.test.ts +++ b/apps/server/src/scheduledTasks/ScheduledTaskService.schedule.test.ts @@ -1,11 +1,18 @@ import * as Scheduler from "../scheduling/Scheduler.ts"; import * as NodeCrypto from "@effect/platform-node/NodeCrypto"; import { expect, it } from "@effect/vitest"; -import { ScheduledTaskUpsertInput } from "@t3tools/contracts"; +import { + type OrchestrationV2ConversationMessage, + type OrchestrationV2Run, + MessageId, + RunId, + ScheduledTaskUpsertInput, +} from "@t3tools/contracts"; import * as DateTime from "effect/DateTime"; import * as Deferred from "effect/Deferred"; import * as Effect from "effect/Effect"; import * as Layer from "effect/Layer"; +import * as Ref from "effect/Ref"; import * as Schema from "effect/Schema"; import * as TestClock from "effect/testing/TestClock"; @@ -173,6 +180,157 @@ it.effect("keeps the task list order when a toggle or edit rewrites updated_at", }).pipe(Effect.provide(SqlitePersistence.layerMemory)), ); +it.effect("keeps one queued occurrence of a recurring task while its bound thread is busy", () => + Effect.gen(function* () { + yield* TestClock.setTime(Date.parse("2026-07-01T09:00:00.000Z")); + // The bound thread's own run never finishes, so every scheduled send lands + // in its queue and stays there. + const activeRun = { + id: RunId.make("run:busy"), + status: "running", + userMessageId: MessageId.make("message:busy"), + } as OrchestrationV2Run; + const thread = yield* Ref.make<{ + readonly runs: ReadonlyArray; + readonly messages: ReadonlyArray; + }>({ runs: [activeRun], messages: [] }); + const runDueWork = yield* Ref.make>(Effect.void); + const layerDependencies = Layer.mergeAll( + NodeCrypto.layer, + ManagedPolicy.layerTest(), + // Capture the poll pass so the test drives each due interval directly. + Layer.mock(Scheduler.Scheduler)({ + register: (name, work) => + name !== "scheduled-tasks" + ? Effect.void + : Effect.context>().pipe( + Effect.flatMap((context) => + Ref.set(runDueWork, work.pipe(Effect.provideContext(context), Effect.ignore)), + ), + ), + }), + Layer.mock(ThreadLaunchService.ThreadLaunchService)({}), + Layer.mock(ManagedProjectFolders.ManagedProjectFolders)({ + namedProjectsRoot: "/projects", + isScratchProject: () => Effect.succeed(false), + }), + Layer.mock(ThreadManagementService.ThreadManagementService)({ + sendToThread: (input) => + Ref.modify(thread, (current) => { + const run = { + id: RunId.make(`run:${input.messageId}`), + status: "queued", + userMessageId: input.messageId, + } as OrchestrationV2Run; + const message = { + id: input.messageId, + threadId: input.threadId, + runId: run.id, + role: "user", + text: input.text, + ...(input.scheduledTaskId === undefined + ? {} + : { scheduledTaskId: input.scheduledTaskId }), + } as OrchestrationV2ConversationMessage; + return [ + { + delivery: "queued", + run, + message, + } as ThreadManagementService.ThreadManagementSendResult, + { runs: [...current.runs, run], messages: [...current.messages, message] }, + ]; + }), + getThreadRecords: () => Ref.get(thread).pipe(Effect.map((records) => records as never)), + getThreadShell: () => + Effect.succeed({ + projectId: "project-busy-thread", + deletedAt: null, + archivedAt: null, + } as never), + }), + Layer.mock(SecretRequests.SecretRequests)({}), + ); + yield* Effect.gen(function* () { + const service = yield* ScheduledTaskService.ScheduledTaskService; + const { task } = yield* service.upsert( + yield* decodeUpsertInput({ + id: "scheduled-task:busy-thread", + title: "Check progress", + prompt: "Check progress", + enabled: true, + schedule: { type: "interval", everyMs: 60_000 }, + projectId: "project-busy-thread", + threadId: "thread:busy", + workspaceStrategy: { type: "root" }, + modelSelection: { instanceId: "codex", model: "gpt-5.4" }, + runtimeMode: "full-access", + interactionMode: "default", + }), + ); + + for (let interval = 0; interval < 3; interval++) { + yield* TestClock.adjust("61 seconds"); + yield* Effect.flatten(Ref.get(runDueWork)); + } + + const queuedOccurrencesOf = (taskId: typeof task.id) => + Ref.get(thread).pipe( + Effect.map( + ({ runs, messages }) => + runs + .filter((run) => run.status === "queued") + .map((run) => messages.find((message) => message.id === run.userMessageId)) + .filter((message) => message?.scheduledTaskId === taskId).length, + ), + ); + const queuedOccurrences = queuedOccurrencesOf(task.id); + expect(yield* queuedOccurrences).toBe(1); + // Skipped intervals still move the schedule forward. + const [listed] = (yield* service.list()).tasks; + expect(Date.parse(listed?.nextRunAt ?? "")).toBeGreaterThan( + Date.parse("2026-07-01T09:03:03.000Z"), + ); + + // Run now is an explicit request, so it still queues behind the held one. + yield* service.runNow({ id: task.id }); + expect(yield* queuedOccurrences).toBe(2); + + // Once the queued copies start, the next interval sends again. + yield* Ref.update(thread, (current) => ({ + ...current, + runs: current.runs.map((run) => + run.status === "queued" ? ({ ...run, status: "running" } as OrchestrationV2Run) : run, + ), + })); + yield* TestClock.adjust("61 seconds"); + yield* Effect.flatten(Ref.get(runDueWork)); + expect(yield* queuedOccurrences).toBe(1); + + // A one-time task's only scheduled occurrence still sends behind a queued Run now. + const { task: once } = yield* service.upsert( + yield* decodeUpsertInput({ + id: "scheduled-task:busy-thread-once", + title: "Check once", + prompt: "Check once", + enabled: true, + schedule: { type: "once", runAt: "2026-07-01T09:05:00.000Z" }, + projectId: "project-busy-thread", + threadId: "thread:busy", + workspaceStrategy: { type: "root" }, + modelSelection: { instanceId: "codex", model: "gpt-5.4" }, + runtimeMode: "full-access", + interactionMode: "default", + }), + ); + yield* service.runNow({ id: once.id }); + yield* TestClock.adjust("61 seconds"); + yield* Effect.flatten(Ref.get(runDueWork)); + expect(yield* queuedOccurrencesOf(once.id)).toBe(2); + }).pipe(Effect.provide(ScheduledTaskService.layer.pipe(Layer.provide(layerDependencies)))); + }).pipe(Effect.provide(SqlitePersistence.layerMemory)), +); + it.effect( "idle polls still wake for tasks saved later, recorded due times, and beside far-future tasks", () => diff --git a/apps/server/src/scheduledTasks/ScheduledTaskService.ts b/apps/server/src/scheduledTasks/ScheduledTaskService.ts index 4fd59e610f0f3..e56f804286ef7 100644 --- a/apps/server/src/scheduledTasks/ScheduledTaskService.ts +++ b/apps/server/src/scheduledTasks/ScheduledTaskService.ts @@ -826,6 +826,24 @@ export const layer = Layer.effect( ), ); + // Reads every run row but only the queued runs' messages, so the cost per + // due interval tracks the thread's run count, not its transcript. + const hasQueuedOccurrence = Effect.fn("ScheduledTaskService.hasQueuedOccurrence")(function* ( + taskId: ScheduledTaskId, + threadId: ThreadId, + ) { + const { runs } = yield* threadManagement.getThreadRecords(threadId, ["runs"]); + const queued = runs.filter((run) => run.status === "queued"); + if (queued.length === 0) return false; + const queuedMessageIds = new Set(queued.map((run) => run.userMessageId)); + const { messages } = yield* threadManagement.getThreadRecords(threadId, ["messages"], { + messageIds: [...queuedMessageIds], + }); + return messages.some( + (message) => queuedMessageIds.has(message.id) && message.scheduledTaskId === taskId, + ); + }); + // The run already started the thread, so a failed write only costs the binding. // A thread deleted before the binding landed already ran its task cleanup, so apply that // cleanup here; a later deletion finds the binding. An unconfirmed lookup keeps the binding. @@ -892,6 +910,19 @@ export const layer = Layer.effect( ) { return { task: active, threadId: null, dispatchFailed: false }; } + // A thread that stays busy would otherwise collect one copy of the + // prompt per interval. A failed read falls through to the normal send. + if ( + trigger === "scheduled" && + active.schedule.type !== "once" && + active.threadId !== null && + (yield* hasQueuedOccurrence(active.id, ThreadId.make(active.threadId)).pipe( + Effect.orElseSucceed(() => false), + )) + ) { + yield* rescheduleMissedRun(active, startedAt, "previous occurrence still queued"); + return { task: active, threadId: null, dispatchFailed: false }; + } // Reject queued deliveries after pause, same-id recreation or trigger changes. if (webhook !== undefined) { const reason = @@ -1114,10 +1145,12 @@ export const layer = Layer.effect( const rescheduleMissedRun = Effect.fn("ScheduledTaskService.rescheduleMissedRun")(function* ( task: ScheduledTask, now: DateTime.DateTime, + reason = "missed", ) { const next = nextRunAt(task, now); - yield* Effect.logInfo("Skipping missed schedule task run", { + yield* Effect.logInfo("Skipping schedule task run", { taskId: task.id, + reason, missedRunAt: task.nextRunAt, rescheduledTo: next, });