From 7068bd081f6ede80a968abdc2e1ffbf0bab44e32 Mon Sep 17 00:00:00 2001 From: Julius Marminge Date: Thu, 22 Oct 2026 12:06:14 +0000 Subject: [PATCH 1/4] test(server): reproduce #16818 recurring tasks queueing duplicate occurrences Co-Authored-By: Claude Opus 5.5 --- .../ScheduledTaskService.schedule.test.ts | 112 +++++++++++++++++- 1 file changed, 111 insertions(+), 1 deletion(-) diff --git a/apps/server/src/scheduledTasks/ScheduledTaskService.schedule.test.ts b/apps/server/src/scheduledTasks/ScheduledTaskService.schedule.test.ts index 09e1d7674044f..828688a70b124 100644 --- a/apps/server/src/scheduledTasks/ScheduledTaskService.schedule.test.ts +++ b/apps/server/src/scheduledTasks/ScheduledTaskService.schedule.test.ts @@ -1,10 +1,17 @@ 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 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"; @@ -167,3 +174,106 @@ it.effect("keeps the task list order when a toggle or edit rewrites updated_at", }).pipe(Effect.provide(ScheduledTaskService.layer.pipe(Layer.provide(layerDependencies)))); }).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, + // 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)), + }), + 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 { runs, messages } = yield* Ref.get(thread); + const queuedOccurrences = runs + .filter((run) => run.status === "queued") + .map((run) => messages.find((message) => message.id === run.userMessageId)) + .filter((message) => message?.scheduledTaskId === task.id) + .map((message) => message?.id); + expect(queuedOccurrences).toHaveLength(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"), + ); + }).pipe(Effect.provide(ScheduledTaskService.layer.pipe(Layer.provide(layerDependencies)))); + }).pipe(Effect.provide(SqlitePersistence.layerMemory)), +); From 9653560c1f57eb412c259a9e04600f0added9a1f Mon Sep 17 00:00:00 2001 From: Julius Marminge Date: Thu, 22 Oct 2026 12:10:09 +0000 Subject: [PATCH 2/4] fix(server): recurring tasks keep one queued occurrence while their thread is busy A timed occurrence of a task bound to a thread now skips, and advances its schedule through the existing missed-run path, while an earlier occurrence of the same task is still queued in that thread. Run now and webhooks still queue. Co-Authored-By: Claude Opus 5.5 --- .../ScheduledTaskService.schedule.test.ts | 21 ++++++++---- .../scheduledTasks/ScheduledTaskService.ts | 34 ++++++++++++++++++- 2 files changed, 47 insertions(+), 8 deletions(-) diff --git a/apps/server/src/scheduledTasks/ScheduledTaskService.schedule.test.ts b/apps/server/src/scheduledTasks/ScheduledTaskService.schedule.test.ts index 828688a70b124..b5d29462d08f1 100644 --- a/apps/server/src/scheduledTasks/ScheduledTaskService.schedule.test.ts +++ b/apps/server/src/scheduledTasks/ScheduledTaskService.schedule.test.ts @@ -262,18 +262,25 @@ it.effect("keeps one queued occurrence of a recurring task while its bound threa yield* Effect.flatten(Ref.get(runDueWork)); } - const { runs, messages } = yield* Ref.get(thread); - const queuedOccurrences = runs - .filter((run) => run.status === "queued") - .map((run) => messages.find((message) => message.id === run.userMessageId)) - .filter((message) => message?.scheduledTaskId === task.id) - .map((message) => message?.id); - expect(queuedOccurrences).toHaveLength(1); + const queuedOccurrences = 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 === task.id).length, + ), + ); + 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); }).pipe(Effect.provide(ScheduledTaskService.layer.pipe(Layer.provide(layerDependencies)))); }).pipe(Effect.provide(SqlitePersistence.layerMemory)), ); diff --git a/apps/server/src/scheduledTasks/ScheduledTaskService.ts b/apps/server/src/scheduledTasks/ScheduledTaskService.ts index 374d6b38027b8..ee59e0257005b 100644 --- a/apps/server/src/scheduledTasks/ScheduledTaskService.ts +++ b/apps/server/src/scheduledTasks/ScheduledTaskService.ts @@ -725,6 +725,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, + ); + }); + const runTask = Effect.fn("ScheduledTaskService.runTask")(function* ( task: ScheduledTask, trigger: "scheduled" | "manual" | "webhook", @@ -782,6 +800,18 @@ export const layer = Layer.effect( ) { return { task: active, threadId: null }; } + // 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.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 }; + } // A queued delivery must not run a task that was paused, deleted and // recreated under the same id, or switched to another trigger while // it waited for its turn. @@ -937,10 +967,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, }); From b18f05c8e24a63c2e582acdb5287cddf97792af2 Mon Sep 17 00:00:00 2001 From: Julius Marminge Date: Thu, 22 Oct 2026 12:27:55 +0000 Subject: [PATCH 3/4] test(server): recurring task sends again once its queued occurrence starts Co-Authored-By: Claude Opus 5.5 --- .../ScheduledTaskService.schedule.test.ts | 11 +++++++++++ 1 file changed, 11 insertions(+) diff --git a/apps/server/src/scheduledTasks/ScheduledTaskService.schedule.test.ts b/apps/server/src/scheduledTasks/ScheduledTaskService.schedule.test.ts index b5d29462d08f1..37d11505d9206 100644 --- a/apps/server/src/scheduledTasks/ScheduledTaskService.schedule.test.ts +++ b/apps/server/src/scheduledTasks/ScheduledTaskService.schedule.test.ts @@ -281,6 +281,17 @@ it.effect("keeps one queued occurrence of a recurring task while its bound threa // 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); }).pipe(Effect.provide(ScheduledTaskService.layer.pipe(Layer.provide(layerDependencies)))); }).pipe(Effect.provide(SqlitePersistence.layerMemory)), ); From fd82defe07917325f46a2b1b05b59da0f99b6d76 Mon Sep 17 00:00:00 2001 From: Julius Marminge Date: Wed, 11 Nov 2026 01:40:49 +0000 Subject: [PATCH 4/4] fix(server): one-time tasks keep their scheduled occurrence behind a queued run The duplicate guard is for recurring schedules. A one-time task's scheduled occurrence is its only one, so skipping it cleared next_run_at for good. AI-assisted; verified with the tests and tools listed above. --- .../ScheduledTaskService.schedule.test.ts | 41 +++++++++++++++---- .../scheduledTasks/ScheduledTaskService.ts | 1 + 2 files changed, 33 insertions(+), 9 deletions(-) diff --git a/apps/server/src/scheduledTasks/ScheduledTaskService.schedule.test.ts b/apps/server/src/scheduledTasks/ScheduledTaskService.schedule.test.ts index c6c27c6ea6ecd..537d071853e34 100644 --- a/apps/server/src/scheduledTasks/ScheduledTaskService.schedule.test.ts +++ b/apps/server/src/scheduledTasks/ScheduledTaskService.schedule.test.ts @@ -274,15 +274,17 @@ it.effect("keeps one queued occurrence of a recurring task while its bound threa yield* Effect.flatten(Ref.get(runDueWork)); } - const queuedOccurrences = 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 === task.id).length, - ), - ); + 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; @@ -304,6 +306,27 @@ it.effect("keeps one queued occurrence of a recurring task while its bound threa 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)), ); diff --git a/apps/server/src/scheduledTasks/ScheduledTaskService.ts b/apps/server/src/scheduledTasks/ScheduledTaskService.ts index d34888328bd85..bbb403d561c6d 100644 --- a/apps/server/src/scheduledTasks/ScheduledTaskService.ts +++ b/apps/server/src/scheduledTasks/ScheduledTaskService.ts @@ -914,6 +914,7 @@ export const layer = Layer.effect( // 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),