From 539ab4c77ad3b731d121cf29685f047d938dcae2 Mon Sep 17 00:00:00 2001 From: JJ Eaton <58495321+oldbettie@users.noreply.github.com> Date: Fri, 13 Nov 2026 10:30:46 +0800 Subject: [PATCH 1/4] feat(server): archive stale scheduled tasks and delete them 30 days later Scheduled tasks that stop doing anything piled up forever. An hourly sweep on the existing scheduler tick now archives a task (turns it off and records why) after 10 days as a fired one-time task, paused task, failing task (3+ consecutive failures over 10 days) or task whose thread was archived, and a never-triggered webhook after 30 days. Archived tasks are deleted 30 days after archiving, with their webhook deliveries. Re-enabling or saving restores one, and a restored webhook keeps its URL. Fork migration 23 adds the archive and failure-counter columns and sets agent tasks that legacy imports left on new_thread to continue. Thread auto-archive now skips threads with an enabled bound schedule, which would otherwise fail every later run. Web and mobile Settings get an Active/Archived filter with the reason, the deletion date and Restore. Co-Authored-By: Claude Opus 5.5 --- .../SettingsScheduledTasksRouteScreen.tsx | 65 ++++-- .../features/settings/scheduledTaskDraft.ts | 3 +- .../settings/scheduledTaskPresentation.ts | 13 ++ ...OrchestratorMcpToolkit.integration.test.ts | 1 + .../ProjectionSettlement.test.ts | 31 ++- .../src/orchestration-v2/ProjectionStore.ts | 5 + apps/server/src/persistence/Migrations.ts | 2 + .../fork/023_ScheduledTaskCleanup.test.ts | 76 +++++++ .../fork/023_ScheduledTaskCleanup.ts | 13 ++ .../scheduledTasks/LegacyScheduledTasks.ts | 2 + .../ScheduledTaskService.profile.test.ts | 1 + .../ScheduledTaskService.test.ts | 201 ++++++++++++++++++ .../scheduledTasks/ScheduledTaskService.ts | 163 +++++++++++++- .../ScheduledTaskService.webhook.test.ts | 58 +++++ .../server/src/scheduledTasks/cleanup.test.ts | 150 +++++++++++++ apps/server/src/scheduledTasks/cleanup.ts | 104 +++++++++ .../chat/ThreadAutomationsPanel.tsx | 3 +- .../ScheduledTasksSettings.browser.test.tsx | 54 ++++- .../settings/ScheduledTasksSettings.tsx | 154 ++++++++++---- .../scheduledTasksSettings.logic.test.ts | 13 ++ .../settings/scheduledTasksSettings.logic.ts | 13 +- apps/web/src/state/scheduledTasks.ts | 5 +- docs/user/project-settings.md | 15 ++ docs/user/thread-sidebar.md | 4 +- packages/client-runtime/package.json | 4 + .../src/scheduledTaskArchive.test.ts | 38 ++++ .../src/scheduledTaskArchive.ts | 38 ++++ packages/contracts/src/scheduledTask.ts | 29 +++ 28 files changed, 1187 insertions(+), 71 deletions(-) create mode 100644 apps/server/src/persistence/Migrations/fork/023_ScheduledTaskCleanup.test.ts create mode 100644 apps/server/src/persistence/Migrations/fork/023_ScheduledTaskCleanup.ts create mode 100644 apps/server/src/scheduledTasks/cleanup.test.ts create mode 100644 apps/server/src/scheduledTasks/cleanup.ts create mode 100644 packages/client-runtime/src/scheduledTaskArchive.test.ts create mode 100644 packages/client-runtime/src/scheduledTaskArchive.ts diff --git a/apps/mobile/src/features/settings/SettingsScheduledTasksRouteScreen.tsx b/apps/mobile/src/features/settings/SettingsScheduledTasksRouteScreen.tsx index 44f1277ebb63c..8c56f3b7894c5 100644 --- a/apps/mobile/src/features/settings/SettingsScheduledTasksRouteScreen.tsx +++ b/apps/mobile/src/features/settings/SettingsScheduledTasksRouteScreen.tsx @@ -89,8 +89,10 @@ import { import { appAtomRegistry } from "../../state/atom-registry"; import { formatNextScheduledTaskRun, + formatScheduledTaskArchive, formatScheduledTaskInterval, } from "./scheduledTaskPresentation"; +import { scheduledTaskRestoreNeedsEditor } from "@t3tools/client-runtime/scheduled-task-archive"; type ScheduledTaskRoutes = { SettingsScheduledTaskNew: undefined; @@ -414,6 +416,7 @@ export function SettingsScheduledTasksRouteScreen() { const navigation = useNavigation>(); const insets = useSafeAreaInsets(); const environmentMenu = useSettingsEnvironmentMenu(); + const [view, setView] = useState<"active" | "archived">("active"); const newTask = () => { if (visibleEnvironments.length === 0) return; resetEditor(); @@ -442,10 +445,19 @@ export function SettingsScheduledTasksRouteScreen() { contentContainerClassName="gap-5 px-5 pt-4" contentContainerStyle={{ paddingBottom: Math.max(insets.bottom, 18) + 18 }} > + {visibleEnvironments.length > 0 ? ( visibleEnvironments.map((environment) => ( projectIds === null || projectIds.includes(task.projectId), + (task) => + (projectIds === null || projectIds.includes(task.projectId)) && + (task.archivedAt != null) === (view === "archived"), ); const canOperate = useAtomValue( serverEnvironment.upsertScheduledTask.permissionAtom(environmentId), @@ -1436,15 +1452,22 @@ function EnvironmentTasks({ } }; - const act = async (task: ScheduledTask, action: "run" | "toggle" | "delete") => { + const act = async (task: ScheduledTask, action: "run" | "toggle" | "restore" | "delete") => { if (!readEnvironmentScope(environmentId, AuthOrchestrationOperateScope)) return; + if (action === "restore" && scheduledTaskRestoreNeedsEditor(task, now)) { + onEdit(task); + return; + } const result = action === "run" ? await runNow({ environmentId, input: { id: task.id } }) - : action === "toggle" + : action === "toggle" || action === "restore" ? await setEnabled({ environmentId, input: { id: task.id, enabled: !task.enabled } }) : await remove({ environmentId, input: { id: task.id } }); - failure(`Could not ${action === "toggle" ? "update" : action} task`, result); + failure( + `Could not ${action === "toggle" || action === "restore" ? "update" : action} task`, + result, + ); }; return ( @@ -1466,7 +1489,11 @@ function EnvironmentTasks({ Loading tasks… ) : visibleTasks?.length === 0 ? ( - {projectIds === null ? "No scheduled tasks yet." : "No tasks in this project."} + {view === "archived" + ? "No archived tasks." + : projectIds === null + ? "No scheduled tasks yet." + : "No tasks in this project."} ) : ( visibleTasks?.map((task, index) => ( @@ -1491,11 +1518,13 @@ function EnvironmentTasks({ {describeSchedule(task, tasks.data?.timeZone)} - {!task.enabled - ? " · Paused" - : task.nextRunAt - ? ` · ${formatNextScheduledTaskRun(task.nextRunAt, now)}` - : ""} + {task.archivedAt != null + ? ` · ${formatScheduledTaskArchive(task)}` + : !task.enabled + ? " · Paused" + : task.nextRunAt + ? ` · ${formatNextScheduledTaskRun(task.nextRunAt, now)}` + : ""} {task.blockedByPolicy ? ( @@ -1514,12 +1543,14 @@ function EnvironmentTasks({ ...(task.threadId !== null && scheduledTaskThreadMode(task) !== "new_thread" ? [{ id: "thread", title: "Open thread" }] : []), - { - id: "toggle", - title: task.enabled ? "Pause" : "Resume", - attributes: { disabled: !canOperate }, - }, - ...(task.schedule.type === "webhook" + task.archivedAt != null + ? { id: "restore", title: "Restore", attributes: { disabled: !canOperate } } + : { + id: "toggle", + title: task.enabled ? "Pause" : "Resume", + attributes: { disabled: !canOperate }, + }, + ...(task.schedule.type === "webhook" || task.archivedAt != null ? [] : [{ id: "run", title: "Run now", attributes: { disabled: !canOperate } }]), { @@ -1549,7 +1580,7 @@ function EnvironmentTasks({ onPress: () => void act(task, "delete"), }, ]); - } else if (action === "toggle" || action === "run") { + } else if (action === "toggle" || action === "restore" || action === "run") { void act(task, action); } }} diff --git a/apps/mobile/src/features/settings/scheduledTaskDraft.ts b/apps/mobile/src/features/settings/scheduledTaskDraft.ts index 452c8c7b28e7c..8b859a42b820e 100644 --- a/apps/mobile/src/features/settings/scheduledTaskDraft.ts +++ b/apps/mobile/src/features/settings/scheduledTaskDraft.ts @@ -322,7 +322,8 @@ export function editDraft(task: ScheduledTask): ScheduledTaskDraft { task.workspaceStrategy.type === "existing_worktree" ? task.workspaceStrategy.worktreePath : "", - enabled: task.enabled, + // Saving restores an archived task, so it opens ready to run again. + enabled: task.enabled || task.archivedAt != null, startFromOrigin: task.workspaceStrategy.type === "worktree" ? (task.workspaceStrategy.startFromOrigin ?? false) diff --git a/apps/mobile/src/features/settings/scheduledTaskPresentation.ts b/apps/mobile/src/features/settings/scheduledTaskPresentation.ts index 0fbc2626f1fbb..c5638013a4421 100644 --- a/apps/mobile/src/features/settings/scheduledTaskPresentation.ts +++ b/apps/mobile/src/features/settings/scheduledTaskPresentation.ts @@ -1,3 +1,6 @@ +import { scheduledTaskArchiveLabel } from "@t3tools/client-runtime/scheduled-task-archive"; +import { scheduledTaskDeletesAt, type ScheduledTask } from "@t3tools/contracts"; + const MINUTE = 60_000; const HOUR = 60 * MINUTE; const DAY = 24 * HOUR; @@ -55,3 +58,13 @@ export function formatNextScheduledTaskRun(nextRunAt: string, now: number): stri }); return `Next run ${date} at ${time}`; } + +export function formatScheduledTaskArchive( + task: Pick, +): string { + const label = scheduledTaskArchiveLabel(task); + const deletesAt = scheduledTaskDeletesAt(task); + if (deletesAt === null) return label; + const date = new Date(deletesAt).toLocaleDateString([], { dateStyle: "medium" }); + return `${label} · Deletes on ${date}`; +} diff --git a/apps/server/src/mcp/OrchestratorMcpToolkit.integration.test.ts b/apps/server/src/mcp/OrchestratorMcpToolkit.integration.test.ts index a852839596046..5ceb133f5d914 100644 --- a/apps/server/src/mcp/OrchestratorMcpToolkit.integration.test.ts +++ b/apps/server/src/mcp/OrchestratorMcpToolkit.integration.test.ts @@ -193,6 +193,7 @@ const layerUnusedScheduledTaskStub = Layer.succeed( getWebhookDelivery: () => Effect.die("unused in this test"), triggerWebhook: () => Effect.die("unused in this test"), awaitWebhookDeliveries: Effect.void, + cleanupStale: () => Effect.die("unused in this test"), }), ); diff --git a/apps/server/src/orchestration-v2/ProjectionSettlement.test.ts b/apps/server/src/orchestration-v2/ProjectionSettlement.test.ts index 3af99805db594..1a534cac7217a 100644 --- a/apps/server/src/orchestration-v2/ProjectionSettlement.test.ts +++ b/apps/server/src/orchestration-v2/ProjectionSettlement.test.ts @@ -18,6 +18,7 @@ import { import * as DateTime from "effect/DateTime"; import * as Effect from "effect/Effect"; import * as Layer from "effect/Layer"; +import * as SqlClient from "effect/sql/SqlClient"; import * as SqlitePersistence from "../persistence/Sqlite.ts"; import * as ProjectionStore from "./ProjectionStore.ts"; @@ -531,12 +532,40 @@ it.effect("sql: lists threads settled before the cutoff for auto-archive, oldest lineage: { rootThreadId: parent, parentThreadId: parent, relationshipToParent: "subagent" }, }); yield* createRun(subagent, "completed"); + // A schedule that still runs into the thread keeps it; a paused one does not. + const sql = yield* SqlClient.SqlClient; + const schedule = (taskId: string, threadId: string, enabled: number) => + sql`INSERT INTO scheduled_tasks ${sql.insert({ + task_id: taskId, + title: taskId, + prompt: "Check in", + enabled, + schedule_json: '{"type":"cron","expression":"0 9 1 * *","timezone":"UTC"}', + project_id: "project-1", + thread_id: threadId, + workspace_strategy_json: '{"type":"root"}', + model_selection_json: '{"instanceId":"codex","model":"gpt-5"}', + runtime_mode: "full-access", + interaction_mode: "default", + created_by: "agent", + creation_source: "mcp", + created_at: DateTime.formatIso(old), + updated_at: DateTime.formatIso(old), + next_run_at: null, + last_run_at: null, + last_run_status: "never", + last_run_error: null, + run_count: 0, + })}`; + yield* schedule("monthly", yield* createThread("archive-scheduled", settled(30)), 1); + const pausedSchedule = yield* createThread("archive-paused-schedule", settled(25)); + yield* schedule("paused", pausedSchedule, 0); const settledBefore = DateTime.subtract(now, { days: 10 }); const candidates = yield* store.getAutoArchiveCandidates({ settledBefore, limit: 10 }); assert.deepEqual( candidates.map((thread) => thread.id), - [oldest, parent, child, due], + [oldest, pausedSchedule, parent, child, due], ); assert.deepEqual( (yield* store.getAutoArchiveCandidates({ settledBefore, limit: 1 })).map( diff --git a/apps/server/src/orchestration-v2/ProjectionStore.ts b/apps/server/src/orchestration-v2/ProjectionStore.ts index dbcd437cbbe26..9deebf251cf14 100644 --- a/apps/server/src/orchestration-v2/ProjectionStore.ts +++ b/apps/server/src/orchestration-v2/ProjectionStore.ts @@ -7207,6 +7207,11 @@ export const layer: Layer.Layer = AND COALESCE(json_extract(t.payload_json, '$.lineage.relationshipToParent'), '') != 'subagent' AND COALESCE(json_extract(t.payload_json, '$.lineage.sideChat'), 0) = 0 AND ${noBlockingWork} + -- Archiving would make every later run of a task bound here fail. + AND NOT EXISTS ( + SELECT 1 FROM scheduled_tasks task + WHERE task.thread_id = t.thread_id AND task.enabled = 1 + ) AND NOT EXISTS ( SELECT 1 FROM orchestration_v2_projection_threads child -- Unary + drops thread_id's TEXT affinity so the parent expression index applies. diff --git a/apps/server/src/persistence/Migrations.ts b/apps/server/src/persistence/Migrations.ts index 751ed047b07c8..54b4cf07c056e 100644 --- a/apps/server/src/persistence/Migrations.ts +++ b/apps/server/src/persistence/Migrations.ts @@ -109,6 +109,7 @@ import ForkMigration019 from "./Migrations/fork/019_NetworkAppReplacement.ts"; import ForkMigration020 from "./Migrations/fork/020_RepairStringifiedModelSelectionOptions.ts"; import ForkMigration021 from "./Migrations/fork/021_UpstreamScheduledTaskProfile.ts"; import ForkMigration022 from "./Migrations/fork/022_NetworkContinuity.ts"; +import ForkMigration023 from "./Migrations/fork/023_ScheduledTaskCleanup.ts"; const migrationEntries = [ [1, "OrchestrationEvents", Migration0001], @@ -221,6 +222,7 @@ const forkMigrationEntries = [ [20, "RepairStringifiedModelSelectionOptions", ForkMigration020, 81], [21, "UpstreamScheduledTaskProfile", ForkMigration021, 81], [22, "NetworkContinuity", ForkMigration022, 81], + [23, "ScheduledTaskCleanup", ForkMigration023, 81], ] as const; export const forkMigrationManifest = forkMigrationEntries.map( diff --git a/apps/server/src/persistence/Migrations/fork/023_ScheduledTaskCleanup.test.ts b/apps/server/src/persistence/Migrations/fork/023_ScheduledTaskCleanup.test.ts new file mode 100644 index 0000000000000..71353f5317c3b --- /dev/null +++ b/apps/server/src/persistence/Migrations/fork/023_ScheduledTaskCleanup.test.ts @@ -0,0 +1,76 @@ +import { assert, it } from "@effect/vitest"; +import * as NodeSqliteClient from "@t3tools/shared/nodeSqliteClient"; +import * as Effect from "effect/Effect"; +import * as SqlClient from "effect/sql/SqlClient"; + +import { runMigrations } from "../../Migrations.ts"; + +const row = (id: string, overrides: Record) => ({ + task_id: id, + title: id, + prompt: "Run", + enabled: 1, + schedule_json: '{"type":"interval","everyMs":60000}', + project_id: "project-1", + thread_id: null, + profile_id: null, + workspace_strategy_json: '{"type":"root"}', + model_selection_json: '{"instanceId":"codex","model":"gpt-5"}', + runtime_mode: "full-access", + interaction_mode: "default", + created_by: "user", + creation_source: "web", + created_at: "2026-11-01T00:00:00.000Z", + updated_at: "2026-11-01T00:00:00.000Z", + next_run_at: null, + last_run_at: null, + last_run_status: "never", + last_run_error: null, + run_count: 0, + ...overrides, +}); + +it.effect("adds the cleanup columns, continues profile tasks and keeps webhook tokens", () => + Effect.gen(function* () { + const sql = yield* SqlClient.SqlClient; + yield* runMigrations({ toForkMigrationInclusive: 22 }); + yield* sql`INSERT INTO scheduled_tasks ${sql.insert( + row("agent", { profile_id: "reviewer", thread_id: "thread-1", thread_mode: "new_thread" }), + )}`; + yield* sql`INSERT INTO scheduled_tasks ${sql.insert( + row("fresh", { thread_mode: "new_thread" }), + )}`; + yield* sql`INSERT INTO scheduled_tasks ${sql.insert( + row("hook", { + schedule_json: '{"type":"webhook","signature":null}', + webhook_token: "token-unchanged", + }), + )}`; + yield* runMigrations(); + + const columns = (yield* sql<{ name: string }>`PRAGMA table_info(scheduled_tasks)`).map( + (column) => column.name, + ); + for (const name of ["archived_at", "archive_reason", "failing_since", "consecutive_failures"]) { + assert.include(columns, name); + } + const rows = yield* sql<{ + task_id: string; + thread_mode: string; + webhook_token: string | null; + archived_at: string | null; + consecutive_failures: number; + }>`SELECT task_id, thread_mode, webhook_token, archived_at, consecutive_failures + FROM scheduled_tasks ORDER BY task_id`; + assert.deepEqual( + rows.map((entry) => [entry.task_id, entry.thread_mode, entry.webhook_token]), + [ + ["agent", "continue", null], + ["fresh", "new_thread", null], + ["hook", "new_thread", "token-unchanged"], + ], + ); + assert.isTrue(rows.every((entry) => entry.archived_at === null)); + assert.isTrue(rows.every((entry) => entry.consecutive_failures === 0)); + }).pipe(Effect.provide(NodeSqliteClient.layer({ filename: ":memory:" }))), +); diff --git a/apps/server/src/persistence/Migrations/fork/023_ScheduledTaskCleanup.ts b/apps/server/src/persistence/Migrations/fork/023_ScheduledTaskCleanup.ts new file mode 100644 index 0000000000000..012e46c7c31a9 --- /dev/null +++ b/apps/server/src/persistence/Migrations/fork/023_ScheduledTaskCleanup.ts @@ -0,0 +1,13 @@ +import * as Effect from "effect/Effect"; +import * as SqlClient from "effect/sql/SqlClient"; + +/** Columns the stale-task sweep reads and writes, plus the mode legacy profile imports missed. */ +export default Effect.gen(function* () { + const sql = yield* SqlClient.SqlClient; + yield* sql`ALTER TABLE scheduled_tasks ADD COLUMN archived_at TEXT`; + yield* sql`ALTER TABLE scheduled_tasks ADD COLUMN archive_reason TEXT`; + yield* sql`ALTER TABLE scheduled_tasks ADD COLUMN failing_since TEXT`; + yield* sql`ALTER TABLE scheduled_tasks ADD COLUMN consecutive_failures INTEGER NOT NULL DEFAULT 0`; + // An agent task always reuses the thread its first run launched. + yield* sql`UPDATE scheduled_tasks SET thread_mode = 'continue' WHERE profile_id IS NOT NULL AND thread_mode = 'new_thread'`; +}); diff --git a/apps/server/src/scheduledTasks/LegacyScheduledTasks.ts b/apps/server/src/scheduledTasks/LegacyScheduledTasks.ts index 639cebec620ac..bcff2b1e1ac1b 100644 --- a/apps/server/src/scheduledTasks/LegacyScheduledTasks.ts +++ b/apps/server/src/scheduledTasks/LegacyScheduledTasks.ts @@ -107,6 +107,8 @@ export const importLegacyScheduledTasks = Effect.fn("importLegacyScheduledTasks" project_id: row.project_id, thread_id: row.thread_id, profile_id: row.profile_id, + // Every fork V1 task runs as an agent, which reuses the thread its first run launched. + thread_mode: "continue", workspace_strategy_json: encodeWorkspaceStrategy({ type: "root" }), model_selection_json: encodeModelSelection(profile?.modelSelection ?? fallbackModel), runtime_mode: runtimeMode, diff --git a/apps/server/src/scheduledTasks/ScheduledTaskService.profile.test.ts b/apps/server/src/scheduledTasks/ScheduledTaskService.profile.test.ts index 3ee2ebd02c691..cc49880b1442b 100644 --- a/apps/server/src/scheduledTasks/ScheduledTaskService.profile.test.ts +++ b/apps/server/src/scheduledTasks/ScheduledTaskService.profile.test.ts @@ -195,6 +195,7 @@ it.effect("imports fork V1 scheduled tasks into the V2 table and drops the legac }); assert.equal(task?.profileId, "reviewer"); assert.equal(task?.threadId, "thread-v1"); + assert.equal(task?.threadMode, "continue"); assert.equal(task?.lastRunStatus, "succeeded"); assert.equal(task?.runCount, 4); assert.equal(task?.nextRunAt, "2026-09-26T00:00:00.000Z"); diff --git a/apps/server/src/scheduledTasks/ScheduledTaskService.test.ts b/apps/server/src/scheduledTasks/ScheduledTaskService.test.ts index 5efa4ea8fe310..ee72b0f35c045 100644 --- a/apps/server/src/scheduledTasks/ScheduledTaskService.test.ts +++ b/apps/server/src/scheduledTasks/ScheduledTaskService.test.ts @@ -926,3 +926,204 @@ it.effect("a run interrupted right after it reserves the task does not block the }).pipe(Effect.provide(layer)); }).pipe(Effect.provide(SqlitePersistence.layerMemory)), ); + +const DAY_MS = 86_400_000; +const CLEANUP_T0 = Date.parse("2026-11-01T00:00:00.000Z"); +const cleanupProject = ProjectId.make("project:test"); + +const cleanupLayer = (mocks: { + readonly launch?: ThreadLaunchService.ThreadLaunchService["Service"]["launch"]; + readonly getThreadShell?: ThreadManagementService.ThreadManagementService["Service"]["getThreadShell"]; +}) => + ScheduledTaskService.layer.pipe( + Layer.provide( + Layer.mergeAll( + ScheduledTaskTestkit.layerLocalProjects(), + ManagedPolicy.layerTest(), + Layer.mock(ThreadLaunchService.ThreadLaunchService)( + mocks.launch === undefined ? {} : { launch: mocks.launch }, + ), + Layer.mock(ThreadManagementService.ThreadManagementService)( + mocks.getThreadShell === undefined ? {} : { getThreadShell: mocks.getThreadShell }, + ), + Layer.mock(SecretRequests.SecretRequests)({}), + Layer.mock(ManagedProjectFolders.ManagedProjectFolders)({ + namedProjectsRoot: "/projects", + isScratchProject: () => Effect.succeed(false), + }), + NodeCrypto.layer, + Layer.succeed(Scheduler.Scheduler, { register: () => Effect.void }), + ), + ), + ); + +const cleanupInput = (id: string, overrides: Partial = {}) => + ({ + id: ScheduledTaskId.make(id), + title: id, + prompt: "Check in", + enabled: true, + schedule: { type: "interval", everyMs: 3_600_000 }, + projectId: cleanupProject, + workspaceStrategy: { type: "root" }, + modelSelection: { instanceId: ProviderInstanceId.make("codex"), model: "gpt-5.4" }, + runtimeMode: "full-access", + interactionMode: "default", + ...overrides, + }) satisfies ScheduledTaskUpsertInput; + +const cleanupColumns = (id: string) => + Effect.gen(function* () { + const sql = yield* SqlClient.SqlClient; + const [row] = yield* sql<{ + archived_at: string | null; + archive_reason: string | null; + failing_since: string | null; + consecutive_failures: number; + }>`SELECT archived_at, archive_reason, failing_since, consecutive_failures + FROM scheduled_tasks WHERE task_id = ${id}`; + return row; + }); + +it.effect("the sweep archives stale tasks, and restoring or editing one clears the archive", () => + Effect.gen(function* () { + yield* TestClock.setTime(CLEANUP_T0); + const boundThread = ThreadId.make("thread:bound"); + const threadArchived = yield* Ref.make(false); + const layer = cleanupLayer({ + getThreadShell: () => + Ref.get(threadArchived).pipe( + Effect.map((archived) => + threadShell(cleanupProject, { + archivedAt: archived ? (DateTime.makeUnsafe(CLEANUP_T0) as never) : null, + }), + ), + ), + }); + yield* Effect.gen(function* () { + const service = yield* ScheduledTaskService.ScheduledTaskService; + yield* service.upsert(cleanupInput("task:paused", { enabled: false })); + yield* service.upsert(cleanupInput("task:active")); + yield* TestClock.setTime(CLEANUP_T0 + 5 * DAY_MS); + yield* service.upsert(cleanupInput("task:bound", { enabled: false, threadId: boundThread })); + yield* Ref.set(threadArchived, true); + + yield* TestClock.setTime(CLEANUP_T0 + 10 * DAY_MS - 1); + yield* service.cleanupStale(); + assert.isTrue((yield* service.list()).tasks.every((task) => task.archivedAt === undefined)); + + yield* TestClock.setTime(CLEANUP_T0 + 10 * DAY_MS); + yield* service.cleanupStale(); + const archivedAt = DateTime.formatIso(DateTime.makeUnsafe(CLEANUP_T0 + 10 * DAY_MS)); + const byId = new Map((yield* service.list()).tasks.map((task) => [task.id, task])); + assert.deepEqual( + [...byId.values()] + .map((task) => [task.id, task.enabled, task.archivedAt, task.archiveReason]) + .sort(), + [ + ["task:active", true, undefined, undefined], + ["task:bound", false, archivedAt, "thread_archived"], + ["task:paused", false, archivedAt, "paused"], + ], + ); + + const restored = yield* service.setEnabled({ + id: ScheduledTaskId.make("task:paused"), + enabled: true, + }); + assert.isTrue(restored.task.enabled); + assert.isUndefined(restored.task.archivedAt); + assert.isUndefined(restored.task.archiveReason); + const edited = yield* service.upsert( + cleanupInput("task:bound", { title: "Renamed", enabled: false, threadId: boundThread }), + ); + assert.isUndefined(edited.task.archivedAt); + assert.isNull((yield* cleanupColumns("task:bound"))?.archived_at); + }).pipe(Effect.provide(layer)); + }).pipe(Effect.provide(SqlitePersistence.layerMemory)), +); + +it.effect( + "counts consecutive failed runs, resets them on success and archives a failing task", + () => + Effect.gen(function* () { + yield* TestClock.setTime(CLEANUP_T0); + const fail = yield* Ref.make(true); + const layer = cleanupLayer({ + launch: () => + Ref.get(fail).pipe( + Effect.flatMap((failing) => + failing + ? Effect.die(new Error("launch failed")) + : Effect.succeed({ threadId: ThreadId.make("thread:ok"), resumed: false } as never), + ), + ), + }); + yield* Effect.gen(function* () { + const service = yield* ScheduledTaskService.ScheduledTaskService; + const id = ScheduledTaskId.make("task:failing"); + yield* service.upsert(cleanupInput(id)); + const runAt = (days: number) => + TestClock.setTime(CLEANUP_T0 + days * DAY_MS).pipe( + Effect.andThen(service.runNow({ id })), + ); + yield* runAt(0); + yield* runAt(1); + assert.deepEqual(yield* cleanupColumns(id), { + archived_at: null, + archive_reason: null, + failing_since: DateTime.formatIso(DateTime.makeUnsafe(CLEANUP_T0)), + consecutive_failures: 2, + }); + yield* Ref.set(fail, false); + yield* runAt(2); + assert.equal((yield* cleanupColumns(id))?.consecutive_failures, 0); + assert.isNull((yield* cleanupColumns(id))?.failing_since); + + yield* Ref.set(fail, true); + yield* runAt(3); + yield* runAt(8); + yield* runAt(12); + yield* service.cleanupStale(); + assert.isNull((yield* cleanupColumns(id))?.archived_at); + yield* TestClock.setTime(CLEANUP_T0 + 13 * DAY_MS); + yield* service.cleanupStale(); + assert.equal((yield* cleanupColumns(id))?.archive_reason, "failing"); + + yield* service.setEnabled({ id, enabled: true }); + assert.deepEqual(yield* cleanupColumns(id), { + archived_at: null, + archive_reason: null, + failing_since: null, + consecutive_failures: 0, + }); + }).pipe(Effect.provide(layer)); + }).pipe(Effect.provide(SqlitePersistence.layerMemory)), +); + +it.effect("the sweep skips a task while it runs", () => + Effect.gen(function* () { + yield* TestClock.setTime(CLEANUP_T0); + const started = yield* Deferred.make(); + const release = yield* Deferred.make(); + const layer = cleanupLayer({ + launch: () => + Deferred.succeed(started, undefined).pipe( + Effect.andThen(Deferred.await(release)), + Effect.as({ threadId: ThreadId.make("thread:slow"), resumed: false } as never), + ), + }); + yield* Effect.gen(function* () { + const service = yield* ScheduledTaskService.ScheduledTaskService; + const id = ScheduledTaskId.make("task:running"); + yield* service.upsert(cleanupInput(id, { enabled: false })); + yield* TestClock.setTime(CLEANUP_T0 + 11 * DAY_MS); + const run = yield* service.runNow({ id }).pipe(Effect.forkChild); + yield* Deferred.await(started); + yield* service.cleanupStale(); + assert.isNull((yield* cleanupColumns(id))?.archived_at); + yield* Deferred.succeed(release, undefined); + yield* Fiber.join(run); + }).pipe(Effect.provide(layer)); + }).pipe(Effect.provide(SqlitePersistence.layerMemory)), +); diff --git a/apps/server/src/scheduledTasks/ScheduledTaskService.ts b/apps/server/src/scheduledTasks/ScheduledTaskService.ts index 1efe55e3187f7..89b36b6de0e83 100644 --- a/apps/server/src/scheduledTasks/ScheduledTaskService.ts +++ b/apps/server/src/scheduledTasks/ScheduledTaskService.ts @@ -62,6 +62,7 @@ import * as ManagedPolicy from "../policy/ManagedPolicy.ts"; import * as DueHint from "../scheduling/DueHint.ts"; import * as Scheduler from "../scheduling/Scheduler.ts"; import { ServerSettingsService } from "../serverSettings.ts"; +import { cleanupDecision } from "./cleanup.ts"; import { importLegacyScheduledTasks } from "./LegacyScheduledTasks.ts"; import { isMissedFixedTimeRun, @@ -101,6 +102,7 @@ const WEBHOOK_DELIVERY_LOG_PROMPT_LIMIT = 64 * 1024; const WEBHOOK_MAX_QUEUED_PER_TASK = 20; // Webhook URLs are reachable by anyone who can reach this machine's address. const WEBHOOK_RATE_LIMIT_PER_MINUTE = 60; +const CLEANUP_INTERVAL_MS = 3_600_000; const LEGACY_DELIVERY_OUTCOMES: ReadonlySet = new Set([ "accepted", "dispatch_failed", @@ -243,6 +245,10 @@ interface ScheduledTaskRow { readonly run_count: number; readonly webhook_token: string | null; readonly webhook_secret: string | null; + readonly archived_at: string | null; + readonly archive_reason: string | null; + readonly failing_since: string | null; + readonly consecutive_failures: number; } interface WebhookDeliveryRow { @@ -306,6 +312,8 @@ export class ScheduledTaskService extends Context.Service< ) => Effect.Effect; /** Waits until every accepted webhook delivery has reached an outcome. */ readonly awaitWebhookDeliveries: Effect.Effect; + /** Archives stale tasks and deletes those archived past retention; runs hourly on the scheduler. */ + readonly cleanupStale: () => Effect.Effect; } >()("t3/scheduledTasks/ScheduledTaskService") {} @@ -425,6 +433,8 @@ const decodeRow = Effect.fnUntraced( lastRunError: row.last_run_error, runCount: row.run_count, ...(webhook === undefined ? {} : { webhook }), + ...(row.archived_at === null ? {} : { archivedAt: row.archived_at }), + ...(row.archive_reason === null ? {} : { archiveReason: row.archive_reason }), }); }, (effect, row, _origin?) => @@ -555,7 +565,11 @@ export const layer = Layer.effect( last_run_error, run_count, webhook_token, - webhook_secret + webhook_secret, + archived_at, + archive_reason, + failing_since, + consecutive_failures FROM scheduled_tasks -- Not updated_at: toggles, runs and edits rewrite it, and clients render -- this order as-is, so rows would move under the pointer. @@ -598,7 +612,11 @@ export const layer = Layer.effect( last_run_error, run_count, webhook_token, - webhook_secret + webhook_secret, + archived_at, + archive_reason, + failing_since, + consecutive_failures FROM scheduled_tasks WHERE task_id = ${id} `.pipe(NodeSqliteClient.readOnly); @@ -750,7 +768,18 @@ export const layer = Layer.effect( webhook_secret = CASE WHEN ${webhook.secretChanged ? 1 : 0} = 1 THEN excluded.webhook_secret ELSE scheduled_tasks.webhook_secret - END + END, + -- Saving restores an archived task; its old failures no longer count against it. + consecutive_failures = CASE + WHEN scheduled_tasks.archived_at IS NULL THEN scheduled_tasks.consecutive_failures + ELSE 0 + END, + failing_since = CASE + WHEN scheduled_tasks.archived_at IS NULL THEN scheduled_tasks.failing_since + ELSE NULL + END, + archived_at = NULL, + archive_reason = NULL RETURNING task_id `.pipe( Effect.mapError((cause) => @@ -811,7 +840,12 @@ export const layer = Layer.effect( next_run_at = ${input.nextRunAtIso}, last_run_status = ${input.status}, last_run_error = ${input.error}, - run_count = run_count + 1 + run_count = run_count + 1, + consecutive_failures = CASE WHEN ${input.status} = 'failed' THEN consecutive_failures + 1 ELSE 0 END, + failing_since = CASE + WHEN ${input.status} = 'failed' THEN COALESCE(failing_since, ${input.startedAtIso}) + ELSE NULL + END WHERE task_id = ${input.id} AND last_run_status = 'running' AND last_run_at = ${input.startedAtIso} @@ -844,6 +878,8 @@ export const layer = Layer.effect( enabled = CASE WHEN ${usedUpOneTime(source, nextRunAfterRun({ ...source, lastRunAt: startedAtIso }, now)) ? 1 : 0} = 1 THEN 0 ELSE enabled END, next_run_at = ${nextRunAfterRun({ ...source, lastRunAt: startedAtIso }, now)}, updated_at = ${iso(now)}, + consecutive_failures = consecutive_failures + 1, + failing_since = COALESCE(failing_since, ${startedAtIso}), run_count = run_count + CASE WHEN EXISTS ( SELECT 1 FROM scheduled_task_webhook_pending WHERE delivery_id = ${pendingDeliveryId ?? null} AND state = 'running' @@ -1306,6 +1342,8 @@ export const layer = Layer.effect( enabled = CASE WHEN ${usedUpOneTime(decoded.success, nextRunAfterRun(decoded.success, now)) ? 1 : 0} = 1 THEN 0 ELSE enabled END, next_run_at = ${nextRunAfterRun(decoded.success, now)}, updated_at = ${iso(now)}, + consecutive_failures = consecutive_failures + 1, + failing_since = COALESCE(failing_since, last_run_at, ${iso(now)}), run_count = run_count + CASE WHEN EXISTS ( SELECT 1 FROM scheduled_task_webhook_pending WHERE task_id IS ${row.task_id} AND state = 'running' @@ -1324,6 +1362,8 @@ export const layer = Layer.effect( SET last_run_status = 'failed', last_run_error = 'Run was interrupted by a server restart.', updated_at = ${iso(now)}, + consecutive_failures = consecutive_failures + 1, + failing_since = COALESCE(failing_since, last_run_at, ${iso(now)}), run_count = run_count + CASE WHEN EXISTS ( SELECT 1 FROM scheduled_task_webhook_pending WHERE task_id IS ${row.task_id} AND state = 'running' @@ -1633,7 +1673,17 @@ export const layer = Layer.effect( UPDATE scheduled_tasks SET enabled = ${input.enabled ? 1 : 0}, next_run_at = ${next}, - updated_at = ${iso(now)} + updated_at = ${iso(now)}, + consecutive_failures = CASE + WHEN ${input.enabled ? 1 : 0} = 1 AND archived_at IS NOT NULL THEN 0 + ELSE consecutive_failures + END, + failing_since = CASE + WHEN ${input.enabled ? 1 : 0} = 1 AND archived_at IS NOT NULL THEN NULL + ELSE failing_since + END, + archived_at = CASE WHEN ${input.enabled ? 1 : 0} = 1 THEN NULL ELSE archived_at END, + archive_reason = CASE WHEN ${input.enabled ? 1 : 0} = 1 THEN NULL ELSE archive_reason END WHERE task_id = ${input.id} RETURNING task_id `.pipe( @@ -1645,8 +1695,14 @@ export const layer = Layer.effect( return yield* taskError("Schedule task not found.", { taskId: input.id }); } yield* notifyChanged; + const { archivedAt: _archivedAt, archiveReason: _archiveReason, ...unarchived } = existing; return { - task: { ...existing, enabled: input.enabled, nextRunAt: next, updatedAt: iso(now) }, + task: { + ...(input.enabled ? unarchived : existing), + enabled: input.enabled, + nextRunAt: next, + updatedAt: iso(now), + }, }; }); @@ -2710,6 +2766,100 @@ export const layer = Layer.effect( }); yield* scheduler.register("webhook-coalescing", dispatchPending()); + const cleanupStale = Effect.fn("ScheduledTaskService.cleanupStale")( + function* () { + const now = yield* localNow; + const nowIso = iso(now); + const rows = yield* selectAllRows(); + const deliveries = yield* sql<{ task_id: string; received_at: string }>` + SELECT task_id, MAX(received_at) AS received_at + FROM scheduled_task_webhook_deliveries GROUP BY task_id + `.pipe(NodeSqliteClient.readOnly); + const latestDeliveryAt = new Map(deliveries.map((row) => [row.task_id, row.received_at])); + const running = yield* Ref.get(activeRuns); + let changed = false; + for (const row of rows) { + const decoded = yield* Effect.result(decodeRow(row)); + if (Result.isFailure(decoded)) continue; + const task = decoded.success; + const shell = + task.threadId === null || task.profileId !== undefined || row.archived_at !== null + ? null + : yield* threadManagement + .getThreadShell(ThreadId.make(task.threadId)) + .pipe(Effect.orElseSucceed(() => null)); + const decision = cleanupDecision( + { + ...task, + archivedAt: row.archived_at, + failingSince: row.failing_since, + consecutiveFailures: row.consecutive_failures, + }, + { + nowMs: DateTime.toEpochMillis(now), + running: running.has(task.id), + threadArchivedAt: shell?.archivedAt == null ? null : iso(shell.archivedAt), + latestDeliveryAt: latestDeliveryAt.get(row.task_id) ?? null, + }, + ); + if (decision._tag === "archive") { + // The updated_at guard drops the decision if an edit or a run landed since the read. + const archived = yield* sql` + UPDATE scheduled_tasks + SET enabled = 0, next_run_at = NULL, + archived_at = ${nowIso}, archive_reason = ${decision.reason} + WHERE task_id = ${row.task_id} AND updated_at = ${row.updated_at} + AND archived_at IS NULL AND last_run_status <> 'running' + RETURNING task_id + `; + if (archived.length > 0) { + changed = true; + yield* Effect.logInfo("Archived stale scheduled task", { + taskId: task.id, + reason: decision.reason, + }); + } + } else if (decision._tag === "delete") { + const deleted = yield* sql.withTransaction( + Effect.gen(function* () { + const removed = yield* sql` + DELETE FROM scheduled_tasks + WHERE task_id = ${row.task_id} AND archived_at = ${row.archived_at} + RETURNING task_id + `; + if (removed.length > 0) { + yield* sql`DELETE FROM scheduled_task_webhook_deliveries WHERE task_id = ${row.task_id}`; + } + return removed.length > 0; + }), + ); + if (deleted) { + changed = true; + yield* forgetWebhookTask(task.id); + yield* Effect.logInfo("Deleted archived scheduled task", { taskId: task.id }); + } + } + } + if (changed) yield* notifyChanged; + }, + Effect.mapError((cause) => + isScheduledTaskError(cause) + ? cause + : taskError("Could not clean up stale schedule tasks.", { cause }), + ), + ); + + const nextCleanupAt = yield* Ref.make(0); + yield* scheduler.register( + "scheduled-task-cleanup", + Effect.gen(function* () { + const nowMs = DateTime.toEpochMillis(yield* DateTime.now); + if (nowMs < (yield* Ref.get(nextCleanupAt))) return; + yield* Ref.set(nextCleanupAt, nowMs + CLEANUP_INTERVAL_MS); + yield* cleanupStale(); + }), + ); + return ScheduledTaskService.of({ list, subscribeList, @@ -2724,6 +2874,7 @@ export const layer = Layer.effect( getWebhookDelivery, triggerWebhook, awaitWebhookDeliveries: FiberSet.awaitEmpty(webhookDeliveries), + cleanupStale, }); }), ); diff --git a/apps/server/src/scheduledTasks/ScheduledTaskService.webhook.test.ts b/apps/server/src/scheduledTasks/ScheduledTaskService.webhook.test.ts index e71ea268f2d40..da8685f55a4b5 100644 --- a/apps/server/src/scheduledTasks/ScheduledTaskService.webhook.test.ts +++ b/apps/server/src/scheduledTasks/ScheduledTaskService.webhook.test.ts @@ -2230,3 +2230,61 @@ it.effect("filters run before replay and coalescing without claiming ignored del { workers }, ); }); + +const CLEANUP_DAY_MS = 86_400_000; + +it.effect("an archived webhook keeps its URL and accepts deliveries again once restored", () => + withService( + ({ service, launches }) => + Effect.gen(function* () { + const { task } = yield* service.upsert(yield* webhookTaskInput()); + yield* TestClock.adjust(29 * CLEANUP_DAY_MS); + yield* service.cleanupStale(); + assert.isUndefined((yield* service.list()).tasks[0]?.archivedAt); + yield* TestClock.adjust(CLEANUP_DAY_MS); + yield* service.cleanupStale(); + const [archived] = (yield* service.list()).tasks; + assert.equal(archived?.archiveReason, "webhook_unused"); + assert.isFalse(archived?.enabled); + assert.equal(archived?.webhook?.path, task.webhook?.path); + assert.equal((yield* service.triggerWebhook(requestFor(task)))._tag, "disabled"); + + const restored = yield* service.setEnabled({ id: task.id, enabled: true }); + assert.isUndefined(restored.task.archivedAt); + assert.equal(restored.task.webhook?.path, task.webhook?.path); + assert.equal((yield* service.triggerWebhook(requestFor(task)))._tag, "accepted"); + yield* Queue.take(launches); + }), + { workers: new Map() }, + ), +); + +it.effect("deletes an archived task and its deliveries 30 days after archiving it", () => + withService( + ({ service, sql }) => + Effect.gen(function* () { + const { task } = yield* service.upsert(yield* webhookTaskInput({ enabled: false })); + // A rejected delivery restarts the unused-webhook clock. + yield* TestClock.adjust(20 * CLEANUP_DAY_MS); + assert.equal((yield* service.triggerWebhook(requestFor(task)))._tag, "disabled"); + yield* TestClock.adjust(20 * CLEANUP_DAY_MS); + yield* service.cleanupStale(); + assert.isUndefined((yield* service.list()).tasks[0]?.archivedAt); + yield* TestClock.adjust(10 * CLEANUP_DAY_MS); + yield* service.cleanupStale(); + assert.equal((yield* service.list()).tasks[0]?.archiveReason, "webhook_unused"); + + yield* TestClock.adjust(30 * CLEANUP_DAY_MS - 1); + yield* service.cleanupStale(); + assert.equal((yield* service.list()).tasks.length, 1); + yield* TestClock.adjust(1); + yield* service.cleanupStale(); + assert.equal((yield* service.list()).tasks.length, 0); + const deliveries = yield* sql` + SELECT 1 FROM scheduled_task_webhook_deliveries WHERE task_id = ${task.id} + `; + assert.equal(deliveries.length, 0); + }), + { workers: new Map() }, + ), +); diff --git a/apps/server/src/scheduledTasks/cleanup.test.ts b/apps/server/src/scheduledTasks/cleanup.test.ts new file mode 100644 index 0000000000000..dc927ea9ca7ff --- /dev/null +++ b/apps/server/src/scheduledTasks/cleanup.test.ts @@ -0,0 +1,150 @@ +import { assert, describe, it } from "@effect/vitest"; +import { ThreadId } from "@t3tools/contracts"; +import * as DateTime from "effect/DateTime"; + +import { type CleanupCandidate, type CleanupContext, cleanupDecision } from "./cleanup.ts"; + +const NOW = Date.parse("2026-11-13T12:00:00.000Z"); +const at = (ms: number) => DateTime.formatIso(DateTime.makeUnsafe(ms)); +const daysAgo = (days: number) => at(NOW - days * 86_400_000); +const inDays = (days: number) => at(NOW + days * 86_400_000); + +const interval = { type: "interval", everyMs: 3_600_000 } as const; +const monthly = { type: "cron", expression: "0 9 1 * *", timezone: "UTC" } as const; +const once = { type: "once", runAt: daysAgo(10) } as const; +const webhook = { type: "webhook", signature: null } as const; + +const task = (overrides: Partial = {}): CleanupCandidate => ({ + enabled: true, + schedule: interval, + runCount: 0, + createdAt: daysAgo(60), + updatedAt: daysAgo(60), + lastRunAt: null, + nextRunAt: inDays(1), + lastRunStatus: "never", + archivedAt: null, + failingSince: null, + consecutiveFailures: 0, + threadId: null, + ...overrides, +}); + +const context = (overrides: Partial = {}): CleanupContext => ({ + nowMs: NOW, + running: false, + threadArchivedAt: null, + latestDeliveryAt: null, + ...overrides, +}); + +const firedOnce = (days: number) => + task({ + enabled: false, + schedule: once, + runCount: 1, + lastRunAt: daysAgo(days), + updatedAt: daysAgo(days), + nextRunAt: null, + lastRunStatus: "succeeded", + }); + +const paused = (days: number) => + task({ enabled: false, nextRunAt: null, runCount: 4, updatedAt: daysAgo(days) }); + +const failing = (days: number, failures: number) => + task({ + runCount: failures, + lastRunStatus: "failed", + failingSince: daysAgo(days), + consecutiveFailures: failures, + }); + +const threadBound = (overrides: Partial = {}) => + task({ + enabled: false, + nextRunAt: null, + threadId: ThreadId.make("thread-1"), + updatedAt: daysAgo(1), + ...overrides, + }); + +const unusedWebhook = (overrides: Partial = {}) => + task({ schedule: webhook, nextRunAt: null, ...overrides }); + +describe("cleanupDecision", () => { + it.each([ + ["1. a one-shot that fired 9 days ago stays", firedOnce(9), context(), "keep"], + ["1. a one-shot that fired 10 days ago is archived", firedOnce(10), context(), "completed"], + ["2. a recurring task paused 9 days ago stays", paused(9), context(), "keep"], + ["2. a recurring task paused 10 days ago is archived", paused(10), context(), "paused"], + ["2. an edit restarts the paused clock", paused(1), context(), "keep"], + [ + "3. an enabled monthly cron whose last run was 25 days ago stays", + task({ schedule: monthly, runCount: 3, lastRunAt: daysAgo(25), updatedAt: daysAgo(25) }), + context(), + "keep", + ], + ["4. three failures over 10 days archive", failing(10, 3), context(), "failing"], + ["4. two failures over 10 days do not", failing(12, 2), context(), "keep"], + ["4. three failures inside 10 days do not", failing(9, 3), context(), "keep"], + [ + "5. a thread task is archived 10 days after its thread", + threadBound(), + context({ threadArchivedAt: daysAgo(10) }), + "thread_archived", + ], + [ + "5. a thread task whose thread was archived 9 days ago stays", + threadBound(), + context({ threadArchivedAt: daysAgo(9) }), + "keep", + ], + [ + "5. an agent task never follows its thread into the archive", + threadBound({ profileId: "agent-1" }), + context({ threadArchivedAt: daysAgo(40) }), + "keep", + ], + [ + "6. a webhook with no deliveries is archived at 30 days", + unusedWebhook(), + context(), + "webhook_unused", + ], + [ + "6. a webhook created 29 days ago stays", + unusedWebhook({ createdAt: daysAgo(29), updatedAt: daysAgo(29) }), + context(), + "keep", + ], + [ + "6. one rejected delivery restarts the webhook clock", + unusedWebhook(), + context({ latestDeliveryAt: daysAgo(5) }), + "keep", + ], + [ + "6. a triggered webhook is never archived", + unusedWebhook({ runCount: 1, lastRunAt: daysAgo(200), lastRunStatus: "succeeded" }), + context({ latestDeliveryAt: daysAgo(200) }), + "keep", + ], + ["10. a running task is skipped", failing(40, 5), context({ running: true }), "keep"], + [ + "10. a task the database marks running is skipped", + { ...paused(40), lastRunStatus: "running" as const }, + context(), + "keep", + ], + ] as const)("%s", (_name, candidate, ctx, expected) => { + const decision = cleanupDecision(candidate, ctx); + assert.equal(decision._tag === "archive" ? decision.reason : decision._tag, expected); + }); + + it("9. archived tasks are deleted 30 days after archive, not before", () => { + const archived = (days: number) => ({ ...paused(90), archivedAt: daysAgo(days) }); + assert.equal(cleanupDecision(archived(29), context())._tag, "keep"); + assert.equal(cleanupDecision(archived(30), context())._tag, "delete"); + }); +}); diff --git a/apps/server/src/scheduledTasks/cleanup.ts b/apps/server/src/scheduledTasks/cleanup.ts new file mode 100644 index 0000000000000..dcbbeb7ddbbcd --- /dev/null +++ b/apps/server/src/scheduledTasks/cleanup.ts @@ -0,0 +1,104 @@ +import { + SCHEDULED_TASK_ARCHIVE_AFTER_DAYS, + SCHEDULED_TASK_ARCHIVE_RETENTION_DAYS, + SCHEDULED_TASK_UNUSED_WEBHOOK_ARCHIVE_AFTER_DAYS, + type ScheduledTask, +} from "@t3tools/contracts"; + +const DAY_MS = 86_400_000; +const FAILURES_BEFORE_ARCHIVE = 3; + +export interface CleanupCandidate extends Pick< + ScheduledTask, + | "enabled" + | "runCount" + | "createdAt" + | "updatedAt" + | "lastRunAt" + | "nextRunAt" + | "lastRunStatus" + | "threadId" + | "profileId" +> { + readonly schedule: { readonly type: ScheduledTask["schedule"]["type"] }; + readonly archivedAt: string | null; + readonly failingSince: string | null; + readonly consecutiveFailures: number; +} + +export interface CleanupContext { + readonly nowMs: number; + /** Whether this process is dispatching the task right now. */ + readonly running: boolean; + /** When the bound thread was archived, for a task bound to an archived thread. */ + readonly threadArchivedAt: string | null; + /** The newest logged webhook delivery of any outcome. */ + readonly latestDeliveryAt: string | null; +} + +export type CleanupDecision = + | { readonly _tag: "keep" } + | { readonly _tag: "delete" } + | { + readonly _tag: "archive"; + readonly reason: "completed" | "paused" | "failing" | "thread_archived" | "webhook_unused"; + }; + +const KEEP: CleanupDecision = { _tag: "keep" }; + +const millis = (value: string | null | undefined) => { + if (value == null) return null; + const parsed = Date.parse(value); + return Number.isFinite(parsed) ? parsed : null; +}; + +/** Archives stale tasks, and deletes archived ones once their retention window has passed. */ +export function cleanupDecision(task: CleanupCandidate, context: CleanupContext): CleanupDecision { + if (context.running || task.lastRunStatus === "running") return KEEP; + const { nowMs } = context; + const olderThan = (at: number | null, days: number) => at !== null && at <= nowMs - days * DAY_MS; + + if (task.archivedAt !== null) { + return olderThan(millis(task.archivedAt), SCHEDULED_TASK_ARCHIVE_RETENTION_DAYS) + ? { _tag: "delete" } + : KEEP; + } + const failing = task.consecutiveFailures > 0; + const nextRunAt = millis(task.nextRunAt); + if (task.enabled && nextRunAt !== null && nextRunAt > nowMs && !failing) return KEEP; + + // An agent task launches a new thread when its thread is archived, so only other tasks follow. + if ( + task.profileId === undefined && + task.threadId !== null && + olderThan(millis(context.threadArchivedAt), SCHEDULED_TASK_ARCHIVE_AFTER_DAYS) + ) { + return { _tag: "archive", reason: "thread_archived" }; + } + if ( + task.consecutiveFailures >= FAILURES_BEFORE_ARCHIVE && + olderThan(millis(task.failingSince), SCHEDULED_TASK_ARCHIVE_AFTER_DAYS) + ) { + return { _tag: "archive", reason: "failing" }; + } + if (task.schedule.type === "webhook" && task.runCount === 0) { + // Quiet is normal for an event trigger, so only one that never fired is stale. + const lastTouched = Math.max( + ...[task.createdAt, task.updatedAt, context.latestDeliveryAt].flatMap( + (value) => millis(value) ?? [], + ), + ); + return olderThan(lastTouched, SCHEDULED_TASK_UNUSED_WEBHOOK_ARCHIVE_AFTER_DAYS) + ? { _tag: "archive", reason: "webhook_unused" } + : KEEP; + } + if (task.enabled) return KEEP; + if (task.schedule.type === "once" && task.runCount > 0) { + return olderThan(millis(task.lastRunAt), SCHEDULED_TASK_ARCHIVE_AFTER_DAYS) + ? { _tag: "archive", reason: "completed" } + : KEEP; + } + return olderThan(millis(task.updatedAt), SCHEDULED_TASK_ARCHIVE_AFTER_DAYS) + ? { _tag: "archive", reason: "paused" } + : KEEP; +} diff --git a/apps/web/src/components/chat/ThreadAutomationsPanel.tsx b/apps/web/src/components/chat/ThreadAutomationsPanel.tsx index 1bec327103211..f62609eec0391 100644 --- a/apps/web/src/components/chat/ThreadAutomationsPanel.tsx +++ b/apps/web/src/components/chat/ThreadAutomationsPanel.tsx @@ -17,6 +17,7 @@ import { import { ThreadDetailsSection } from "./ThreadDetailsSection"; import { cn } from "../../lib/utils"; import { ScheduledRelativeLabel } from "../settings/ScheduledTasksSettings"; +import { isScheduledTaskArchived } from "@t3tools/client-runtime/scheduled-task-archive"; import { scheduleLabel } from "../settings/scheduledTasksSettings.logic"; import { useEnvironmentQuery } from "../../state/query"; import { serverEnvironment } from "../../state/server"; @@ -78,7 +79,7 @@ export function ThreadAutomationsPanel(props: { const [busyTaskId, setBusyTaskId] = useState(null); const boundTasks = (tasksQuery.data?.tasks ?? []).filter( - (task) => task.threadId === props.threadId, + (task) => task.threadId === props.threadId && !isScheduledTaskArchived(task), ); if (tasksQuery.error === null && boundTasks.length === 0) return null; diff --git a/apps/web/src/components/settings/ScheduledTasksSettings.browser.test.tsx b/apps/web/src/components/settings/ScheduledTasksSettings.browser.test.tsx index ca8192dd894e3..7d637600ebc8b 100644 --- a/apps/web/src/components/settings/ScheduledTasksSettings.browser.test.tsx +++ b/apps/web/src/components/settings/ScheduledTasksSettings.browser.test.tsx @@ -47,7 +47,22 @@ const threadFamilyRead = { threads: [{ id: task.threadId, title: "Fix login bug" }], }; -const state = vi.hoisted(() => ({ connected: true, tasksLoaded: true, secondary: false })); +const archivedTask: ScheduledTask = { + ...task, + id: ScheduledTaskId.make("stale"), + title: "Stale sweep", + enabled: false, + archivedAt: "2026-11-01T12:00:00.000Z", + archiveReason: "paused", +}; + +const state = vi.hoisted(() => ({ + connected: true, + tasksLoaded: true, + secondary: false, + archived: false, + commands: [] as unknown[], +})); const secondaryId = EnvironmentId.make("secondary"); const permissions = new Map>(); @@ -55,6 +70,8 @@ function setup() { state.connected = true; state.tasksLoaded = true; state.secondary = false; + state.archived = false; + state.commands = []; permissions.clear(); permissions.set(environmentId, Atom.make(true)); permissions.set(secondaryId, Atom.make(true)); @@ -125,12 +142,23 @@ vi.mock(import("../../state/query"), async (original) => ({ data: request === null || request === threadFamilyRead ? request - : { tasks: state.tasksLoaded ? [task] : [] }, + : { tasks: state.tasksLoaded ? (state.archived ? [task, archivedTask] : [task]) : [] }, error: null, isPending: false, refresh: vi.fn(), })) as never, })); +vi.mock(import("~/state/session"), async (original) => ({ + ...(await original()), + readEnvironmentScope: () => true, +})); +vi.mock(import("../../state/use-atom-command"), async (original) => ({ + ...(await original()), + useAtomCommand: (() => async (value: unknown) => { + state.commands.push(value); + return { _tag: "Success", value: undefined }; + }) as never, +})); vi.mock(import("../../state/environments"), async (original) => ({ ...(await original()), useEnvironment: ((id: EnvironmentId) => @@ -182,6 +210,28 @@ describe("ScheduledTasksSettings", () => { await expect.element(screen.getByRole("button", { name: "New task" })).toBeDisabled(); }); + it("lists an archived task only under Archived and restores it", async () => { + setup(); + state.archived = true; + await using screen = await render(); + await expect.element(screen.getByText("Review issues")).toBeVisible(); + await expect.element(screen.getByText("Stale sweep")).not.toBeInTheDocument(); + + await screen.getByRole("button", { name: /^Archived/ }).click(); + await expect.element(screen.getByText("Stale sweep")).toBeVisible(); + await expect.element(screen.getByText("Review issues")).not.toBeInTheDocument(); + await expect + .element(screen.getByText(/Archived: paused for 10 days · Deletes on/)) + .toBeVisible(); + + await screen.getByRole("button", { name: "Restore" }).click(); + await vi.waitFor(() => + expect(state.commands).toEqual([ + { environmentId, input: { id: archivedTask.id, enabled: true } }, + ]), + ); + }); + it("keeps a linked task's unsaved edits across a disconnect and reconnect", async () => { setup(); await using screen = await render(); diff --git a/apps/web/src/components/settings/ScheduledTasksSettings.tsx b/apps/web/src/components/settings/ScheduledTasksSettings.tsx index 5e03b872443fd..9b9f2e796f769 100644 --- a/apps/web/src/components/settings/ScheduledTasksSettings.tsx +++ b/apps/web/src/components/settings/ScheduledTasksSettings.tsx @@ -2,6 +2,7 @@ import { useAtomValue } from "@effect/atom-react"; import { connectionStatusTitle } from "@t3tools/client-runtime/connection"; import { AsyncResult } from "effect/reactivity"; import { + ArchiveRestoreIcon, ChevronDownIcon, Clock3Icon, CopyIcon, @@ -85,6 +86,13 @@ import { managedCapabilityDenial, } from "@t3tools/contracts"; import { resolvePastedPath } from "@t3tools/shared/path"; +import { + isScheduledTaskArchived, + scheduledTaskArchiveLabel, + scheduledTaskRestoreNeedsEditor, + scheduledTasksInView, + type ScheduledTaskView, +} from "@t3tools/client-runtime/scheduled-task-archive"; import { useManagedPolicy } from "~/state/managedPolicy"; import { ManagedNotice } from "../ManagedPolicy"; import { readEnvironmentScope } from "~/state/session"; @@ -105,6 +113,7 @@ import { scheduleFromDraft, scheduleLabel, scheduledTaskDefaultModel, + scheduledTaskDeletesLabel, WEEKDAY_LABELS, taskToDraft, unbindDraftFromThread, @@ -435,24 +444,57 @@ export function ScheduledTasksSettings(target: { permittedEnvironments.find( (entry) => entry.environmentId === defaultEnvironment?.environmentId, ) ?? permittedEnvironments[0]; + const [view, setView] = useState("active"); + const listQueries = useAtomValue( + scheduledTasksListAtom( + environmentIdsKey(connectedEnvironments.map((entry) => entry.environmentId)), + ), + ); + const archivedCount = listQueries.reduce((count, query, index) => { + const environmentId = connectedEnvironments[index]!.environmentId; + const tasks = AsyncResult.getOrElse(query, () => null)?.tasks ?? []; + return ( + count + + tasks.filter( + (task) => + isScheduledTaskArchived(task) && + matchesScheduledTaskScope(scope, environmentId, task.projectId), + ).length + ); + }, 0); return ( - creationEnvironment && - setEditor({ environmentId: creationEnvironment.environmentId, task: null }) - } - > - - New task - +
+ { + const next = values[0]; + if (next === "active" || next === "archived") setView(next); + }} + > + Active + + Archived{archivedCount > 0 ? ` (${archivedCount})` : ""} + + + +
} > {scope.kind === "unavailable" ? ( @@ -471,6 +513,7 @@ export function ScheduledTasksSettings(target: { ) : environments.length > 1 ? ( ( void; @@ -542,8 +588,10 @@ function ScheduledTaskEnvironmentList({ const linkedState = states.find((state) => state.environment.environmentId === environmentId); const linkedTask = linkedState?.tasks?.find((task) => task.id === taskId); - const loaded = states.filter((state) => state.connected && state.error === null && state.tasks); - const populated = loaded.filter((state) => state.tasks && state.tasks.length > 0); + const loaded = states + .filter((state) => state.connected && state.error === null && state.tasks) + .map((state) => ({ ...state, tasks: scheduledTasksInView(state.tasks ?? [], view) })); + const populated = loaded.filter((state) => state.tasks.length > 0); const summaries: Record = { "Not connected:": [], "Loading scheduled tasks on": [], @@ -560,7 +608,7 @@ function ScheduledTaskEnvironmentList({ summaries["Could not load scheduled tasks:"]!.push(`${label}: ${error}`); } else if (!tasks) { summaries["Loading scheduled tasks on"]!.push(label); - } else if (tasks.length === 0 && populated.length > 0) { + } else if (scheduledTasksInView(tasks, view).length === 0 && populated.length > 0) { summaries["No tasks on"]!.push(label); } } @@ -586,7 +634,7 @@ function ScheduledTaskEnvironmentList({ {connectionStatusTitle(environment.connection)} - {tasks?.map((task) => ( + {tasks.map((task) => ( ) : null} - {populated.length === 0 && loaded.length > 0 ? ( - - ) : null} + {populated.length === 0 && loaded.length > 0 ? : null} {Object.entries(summaries).map(([label, entries]) => { if (entries.length === 0) return null; const summary = `${label} ${entries.join(", ")}`; @@ -653,10 +696,12 @@ function ScheduledTaskEnvironmentSubtitle({ } function ScheduledTaskEnvironmentSection({ + view, environment, taskId, onEdit, }: { + readonly view: ScheduledTaskView; readonly environment: EnvironmentPresentation; readonly taskId?: ScheduledTaskId | undefined; readonly onEdit: (environmentId: EnvironmentId, task: ScheduledTask) => void; @@ -705,13 +750,10 @@ function ScheduledTaskEnvironmentSection({ role="status" /> ) : null} - {tasks.length === 0 ? ( - + {scheduledTasksInView(tasks, view).length === 0 ? ( + ) : ( - tasks.map((task) => ( + scheduledTasksInView(tasks, view).map((task) => ( + ) : ( + + ); +} + function ScheduledTaskRow({ environmentId, task, @@ -744,6 +800,7 @@ function ScheduledTaskRow({ const [busy, setBusy] = useState(false); const [deliveriesOpen, setDeliveriesOpen] = useState(false); const isWebhook = task.schedule.type === "webhook"; + const archived = isScheduledTaskArchived(task); const toggle = useAtomCommand(serverEnvironment.setScheduledTaskEnabled, { label: "scheduled task enabled", }); @@ -753,11 +810,15 @@ function ScheduledTaskRow({ const remove = useAtomCommand(serverEnvironment.deleteScheduledTask, { label: "scheduled task delete", }); - const act = async (action: "toggle" | "run" | "delete") => { + const act = async (action: "toggle" | "restore" | "run" | "delete") => { if (busy || !readEnvironmentScope(environmentId, AuthOrchestrationOperateScope)) return; + if (action === "restore" && scheduledTaskRestoreNeedsEditor(task, Date.now())) { + onEdit(); + return; + } setBusy(true); const result = - action === "toggle" + action === "toggle" || action === "restore" ? await toggle({ environmentId, input: { id: task.id, enabled: !task.enabled } }) : action === "run" ? await run({ environmentId, input: { id: task.id } }) @@ -781,7 +842,12 @@ function ScheduledTaskRow({
{scheduleLabel(task.schedule, timeZone, task.nextRunAt)} ·{" "} - {task.blockedByPolicy !== undefined ? ( + {archived ? ( + <> + {scheduledTaskArchiveLabel(task)} + {scheduledTaskDeletesLabel(task) ? ` · ${scheduledTaskDeletesLabel(task)}` : null} + + ) : task.blockedByPolicy !== undefined ? ( task.blockedByPolicy ) : !task.enabled ? ( "Paused" @@ -806,12 +872,24 @@ function ScheduledTaskRow({ } control={
- void act("toggle")} - /> + {archived ? ( + + ) : ( + void act("toggle")} + /> + )} Deliveries - ) : ( + ) : archived ? null : ( void act("run")}> Run now diff --git a/apps/web/src/components/settings/scheduledTasksSettings.logic.test.ts b/apps/web/src/components/settings/scheduledTasksSettings.logic.test.ts index 73994e1db6baf..64f66ca13663f 100644 --- a/apps/web/src/components/settings/scheduledTasksSettings.logic.test.ts +++ b/apps/web/src/components/settings/scheduledTasksSettings.logic.test.ts @@ -235,6 +235,19 @@ describe("one-time scheduled tasks", () => { }); }); +describe("archived scheduled tasks", () => { + it("opens an archived task enabled, since saving restores it", () => { + const archived: ScheduledTask = { + ...legacyTask, + enabled: false, + archivedAt: "2026-11-01T00:00:00.000Z", + archiveReason: "paused", + }; + expect(taskToDraft(archived).enabled).toBe(true); + expect(taskToDraft({ ...legacyTask, enabled: false }).enabled).toBe(false); + }); +}); + describe("webhook scheduled tasks", () => { const signature = { header: "x-signature", encoding: "base64", prefix: "" } as const; const webhookTask: ScheduledTask = { diff --git a/apps/web/src/components/settings/scheduledTasksSettings.logic.ts b/apps/web/src/components/settings/scheduledTasksSettings.logic.ts index 7e27f08dba794..2ee43f3844332 100644 --- a/apps/web/src/components/settings/scheduledTasksSettings.logic.ts +++ b/apps/web/src/components/settings/scheduledTasksSettings.logic.ts @@ -1,4 +1,5 @@ -import { isWebhookDedupeTemplate } from "@t3tools/contracts"; +import { isWebhookDedupeTemplate, scheduledTaskDeletesAt } from "@t3tools/contracts"; +import { isScheduledTaskArchived } from "@t3tools/client-runtime/scheduled-task-archive"; import { EnvironmentId, type ProjectId, @@ -37,6 +38,8 @@ const ONCE_FORMATTER = new Intl.DateTimeFormat(undefined, { timeStyle: "short", }); +const DATE_FORMATTER = new Intl.DateTimeFormat(undefined, { dateStyle: "medium" }); + export function scheduleLabel( schedule: ScheduledTaskSchedule, timeZone?: string, @@ -62,6 +65,11 @@ export function scheduleLabel( return `${days} at ${schedule.timeOfDay}${zone === undefined ? "" : ` ${zone}`}`; } +export function scheduledTaskDeletesLabel(task: Pick): string | null { + const deletesAt = scheduledTaskDeletesAt(task); + return deletesAt === null ? null : `Deletes on ${DATE_FORMATTER.format(new Date(deletesAt))}`; +} + export function matchesScheduledTaskScope( scope: ResolvedSettingsScope, environmentId: EnvironmentId, @@ -290,7 +298,8 @@ export function taskToDraft(task: ScheduledTask): DraftState { editingId: task.id, title: task.title, prompt: task.prompt, - enabled: task.enabled, + // Saving restores an archived task, so it opens ready to run again. + enabled: task.enabled || isScheduledTaskArchived(task), scheduleMode: schedule.type === "fixed_time" ? "fixed" : schedule.type === "cron" ? "kept" : schedule.type, intervalMinutes: diff --git a/apps/web/src/state/scheduledTasks.ts b/apps/web/src/state/scheduledTasks.ts index 718bfc4e247a8..ce2d747524fba 100644 --- a/apps/web/src/state/scheduledTasks.ts +++ b/apps/web/src/state/scheduledTasks.ts @@ -22,7 +22,10 @@ const agentScheduledTasksAtom = Atom.make((get): ReadonlyArray { + it("lists archived tasks only under Archived", () => { + const active: { id: string; archivedAt?: string } = { id: "active" }; + const archived = { id: "archived", archivedAt: "2026-11-01T00:00:00.000Z" }; + expect(scheduledTasksInView([active, archived], "active")).toEqual([active]); + expect(scheduledTasksInView([active, archived], "archived")).toEqual([archived]); + }); + + it.each([ + ["completed", "Archived: finished its one-time run"], + ["paused", "Archived: paused for 10 days"], + ["failing", "Archived: failing for 10 days"], + ["thread_archived", "Archived: its thread was archived"], + ["webhook_unused", "Archived: never triggered in 30 days"], + ["added_later", "Archived"], + [undefined, "Archived"], + ])("labels reason %s", (archiveReason, label) => { + expect(scheduledTaskArchiveLabel({ archiveReason })).toBe(label); + }); + + it("restores a one-time task whose time passed through the editor", () => { + const now = Date.parse("2026-11-13T00:00:00.000Z"); + const once = (runAt: string) => ({ schedule: { type: "once", runAt } as const }); + expect(scheduledTaskRestoreNeedsEditor(once("2026-11-12T00:00:00.000Z"), now)).toBe(true); + expect(scheduledTaskRestoreNeedsEditor(once("2026-11-14T00:00:00.000Z"), now)).toBe(false); + expect( + scheduledTaskRestoreNeedsEditor({ schedule: { type: "interval", everyMs: 60_000 } }, now), + ).toBe(false); + }); +}); diff --git a/packages/client-runtime/src/scheduledTaskArchive.ts b/packages/client-runtime/src/scheduledTaskArchive.ts new file mode 100644 index 0000000000000..2c4fa5024663a --- /dev/null +++ b/packages/client-runtime/src/scheduledTaskArchive.ts @@ -0,0 +1,38 @@ +import { + SCHEDULED_TASK_ARCHIVE_AFTER_DAYS, + SCHEDULED_TASK_UNUSED_WEBHOOK_ARCHIVE_AFTER_DAYS, + type ScheduledTask, +} from "@t3tools/contracts"; + +export type ScheduledTaskView = "active" | "archived"; + +export function isScheduledTaskArchived(task: Pick): boolean { + return task.archivedAt != null; +} + +export function scheduledTasksInView>( + tasks: ReadonlyArray, + view: ScheduledTaskView, +): T[] { + return tasks.filter((task) => isScheduledTaskArchived(task) === (view === "archived")); +} + +const ARCHIVE_REASON_LABELS: Record = { + completed: "Archived: finished its one-time run", + paused: `Archived: paused for ${SCHEDULED_TASK_ARCHIVE_AFTER_DAYS} days`, + failing: `Archived: failing for ${SCHEDULED_TASK_ARCHIVE_AFTER_DAYS} days`, + thread_archived: "Archived: its thread was archived", + webhook_unused: `Archived: never triggered in ${SCHEDULED_TASK_UNUSED_WEBHOOK_ARCHIVE_AFTER_DAYS} days`, +}; + +export function scheduledTaskArchiveLabel(task: Pick): string { + return (task.archiveReason && ARCHIVE_REASON_LABELS[task.archiveReason]) || "Archived"; +} + +/** A one-time task whose run time passed can't be re-enabled; the editor asks for a new time. */ +export function scheduledTaskRestoreNeedsEditor( + task: Pick, + nowMs: number, +): boolean { + return task.schedule.type === "once" && !(Date.parse(task.schedule.runAt) > nowMs); +} diff --git a/packages/contracts/src/scheduledTask.ts b/packages/contracts/src/scheduledTask.ts index 388f0a20419b3..7701fa9c70dd2 100644 --- a/packages/contracts/src/scheduledTask.ts +++ b/packages/contracts/src/scheduledTask.ts @@ -1,3 +1,5 @@ +import * as DateTime from "effect/DateTime"; +import * as Option from "effect/Option"; import * as Schema from "effect/Schema"; import { @@ -262,6 +264,30 @@ export function scheduledTaskThreadMode( return task.threadMode ?? (task.threadId === null ? "new_thread" : "continue"); } +/** Why the server archived a task. Any string, so a reason added later still decodes. */ +export const ScheduledTaskArchiveReason = Schema.Union([ + Schema.Literals(["completed", "paused", "failing", "thread_archived", "webhook_unused"]), + Schema.String, +]); +export type ScheduledTaskArchiveReason = typeof ScheduledTaskArchiveReason.Type; + +/** Stale tasks are archived after this many days (never-triggered webhooks after the longer window). */ +export const SCHEDULED_TASK_ARCHIVE_AFTER_DAYS = 10; +export const SCHEDULED_TASK_UNUSED_WEBHOOK_ARCHIVE_AFTER_DAYS = 30; +/** Archived tasks are deleted this many days after they were archived. */ +export const SCHEDULED_TASK_ARCHIVE_RETENTION_DAYS = 30; + +/** When the server deletes an archived task, or null for one that is not archived. */ +export function scheduledTaskDeletesAt(task: Pick): string | null { + if (task.archivedAt == null) return null; + return DateTime.make(task.archivedAt).pipe( + Option.map((archivedAt) => + DateTime.formatIso(DateTime.add(archivedAt, { days: SCHEDULED_TASK_ARCHIVE_RETENTION_DAYS })), + ), + Option.getOrNull, + ); +} + export const ScheduledTask = Schema.Struct({ id: ScheduledTaskId, title: TrimmedNonEmptyString, @@ -287,6 +313,9 @@ export const ScheduledTask = Schema.Struct({ lastRunError: Schema.NullOr(Schema.String), runCount: Schema.Int.check(Schema.isGreaterThanOrEqualTo(0)), webhook: Schema.optional(ScheduledTaskWebhookEndpoint), + /** Set when the server archived the stale task; it is disabled and deleted after the retention window. */ + archivedAt: Schema.optional(IsoDateTime), + archiveReason: Schema.optional(ScheduledTaskArchiveReason), /** Why the task cannot run regardless of `enabled`, such as a managed policy refusing webhooks. */ blockedByPolicy: Schema.optionalKey(Schema.String), }); From e1e6d06ee6a89e1f452096d277718e197ae7acd3 Mon Sep 17 00:00:00 2001 From: JJ Eaton <58495321+oldbettie@users.noreply.github.com> Date: Fri, 13 Nov 2026 12:23:41 +0700 Subject: [PATCH 2/4] fix(server): stale-task sweep no longer deletes a task whose run just started The sweep reads task rows and active runs once. A manual run that started while the sweep waited on an earlier row's thread lookup left archived_at unchanged, so the delete removed the running task. The delete now requires the row's read updated_at and a non-running status, and holds the run reservation so a run cannot start mid-delete. Co-Authored-By: Claude Opus 5.5 --- .../ScheduledTaskService.test.ts | 48 ++++++++++++++ .../scheduledTasks/ScheduledTaskService.ts | 63 +++++++++++-------- 2 files changed, 85 insertions(+), 26 deletions(-) diff --git a/apps/server/src/scheduledTasks/ScheduledTaskService.test.ts b/apps/server/src/scheduledTasks/ScheduledTaskService.test.ts index ee72b0f35c045..17d945aad2e44 100644 --- a/apps/server/src/scheduledTasks/ScheduledTaskService.test.ts +++ b/apps/server/src/scheduledTasks/ScheduledTaskService.test.ts @@ -1127,3 +1127,51 @@ it.effect("the sweep skips a task while it runs", () => }).pipe(Effect.provide(layer)); }).pipe(Effect.provide(SqlitePersistence.layerMemory)), ); + +it.effect("the sweep keeps an archived task whose run starts after the sweep read it", () => + Effect.gen(function* () { + yield* TestClock.setTime(CLEANUP_T0); + const sweepPaused = yield* Deferred.make(); + const resumeSweep = yield* Deferred.make(); + const started = yield* Deferred.make(); + const release = yield* Deferred.make(); + const pauseSweep = yield* Ref.make(false); + const layer = cleanupLayer({ + getThreadShell: () => + Effect.gen(function* () { + if (yield* Ref.get(pauseSweep)) { + yield* Deferred.succeed(sweepPaused, undefined); + yield* Deferred.await(resumeSweep); + } + return threadShell(cleanupProject); + }), + launch: () => + Deferred.succeed(started, undefined).pipe( + Effect.andThen(Deferred.await(release)), + Effect.as({ threadId: ThreadId.make("thread:manual"), resumed: false } as never), + ), + }); + yield* Effect.gen(function* () { + const service = yield* ScheduledTaskService.ScheduledTaskService; + const id = ScheduledTaskId.make("task:archived"); + yield* service.upsert(cleanupInput(id, { enabled: false })); + yield* TestClock.setTime(CLEANUP_T0 + 10 * DAY_MS); + yield* service.cleanupStale(); + yield* TestClock.setTime(CLEANUP_T0 + 41 * DAY_MS); + // A bound task read after the archived one pauses the sweep on its thread lookup. + yield* service.upsert( + cleanupInput("task:bound", { threadId: ThreadId.make("thread:bound") }), + ); + yield* Ref.set(pauseSweep, true); + const sweep = yield* service.cleanupStale().pipe(Effect.forkChild); + yield* Deferred.await(sweepPaused); + const run = yield* service.runNow({ id }).pipe(Effect.forkChild); + yield* Deferred.await(started); + yield* Deferred.succeed(resumeSweep, undefined); + yield* Fiber.join(sweep); + assert.isDefined((yield* service.list()).tasks.find((task) => task.id === id)); + yield* Deferred.succeed(release, undefined); + yield* Fiber.join(run); + }).pipe(Effect.provide(layer)); + }).pipe(Effect.provide(SqlitePersistence.layerMemory)), +); diff --git a/apps/server/src/scheduledTasks/ScheduledTaskService.ts b/apps/server/src/scheduledTasks/ScheduledTaskService.ts index 89b36b6de0e83..32871928a6617 100644 --- a/apps/server/src/scheduledTasks/ScheduledTaskService.ts +++ b/apps/server/src/scheduledTasks/ScheduledTaskService.ts @@ -525,6 +525,19 @@ export const layer = Layer.effect( ReadonlyMap> >(new Map()); const activeRuns = yield* Ref.make>(new Set()); + const reserveRun = (id: ScheduledTaskId) => + Ref.modify(activeRuns, (active) => { + if (active.has(id)) return [false, active] as const; + const next = new Set(active); + next.add(id); + return [true, next] as const; + }); + const releaseRun = (id: ScheduledTaskId) => + Ref.update(activeRuns, (active) => { + const next = new Set(active); + next.delete(id); + return next; + }); /** Caller-key claims this process is still carrying from request to finished run. */ const keyedInFlight = yield* Ref.make>(new Set()); // Sliding(1) coalesces the dirty-signal: every notification triggers a @@ -1220,26 +1233,14 @@ export const layer = Layer.effect( // Reserve and release as one bracket so an interrupt cannot leave the task marked busy. return yield* Effect.acquireUseRelease( - Ref.modify(activeRuns, (active) => { - if (active.has(task.id)) return [false, active] as const; - const next = new Set(active); - next.add(task.id); - return [true, next] as const; - }), + reserveRun(task.id), (reserved) => reserved ? run : trigger !== "scheduled" ? Effect.fail(taskError("Schedule task is already running.", { taskId: task.id })) : Effect.succeed({ task, threadId: null, dispatchFailed: false }), - (reserved) => - reserved - ? Ref.update(activeRuns, (active) => { - const next = new Set(active); - next.delete(task.id); - return next; - }) - : Effect.void, + (reserved) => (reserved ? releaseRun(task.id) : Effect.void), ); }); @@ -2820,18 +2821,28 @@ export const layer = Layer.effect( }); } } else if (decision._tag === "delete") { - const deleted = yield* sql.withTransaction( - Effect.gen(function* () { - const removed = yield* sql` - DELETE FROM scheduled_tasks - WHERE task_id = ${row.task_id} AND archived_at = ${row.archived_at} - RETURNING task_id - `; - if (removed.length > 0) { - yield* sql`DELETE FROM scheduled_task_webhook_deliveries WHERE task_id = ${row.task_id}`; - } - return removed.length > 0; - }), + // Holding the run reservation keeps a run from starting mid-delete; the updated_at + // and status guards drop the decision if a run or edit landed since the read. + const deleted = yield* Effect.acquireUseRelease( + reserveRun(task.id), + (reserved) => + reserved + ? sql.withTransaction( + Effect.gen(function* () { + const removed = yield* sql` + DELETE FROM scheduled_tasks + WHERE task_id = ${row.task_id} AND archived_at = ${row.archived_at} + AND updated_at = ${row.updated_at} AND last_run_status <> 'running' + RETURNING task_id + `; + if (removed.length > 0) { + yield* sql`DELETE FROM scheduled_task_webhook_deliveries WHERE task_id = ${row.task_id}`; + } + return removed.length > 0; + }), + ) + : Effect.succeed(false), + (reserved) => (reserved ? releaseRun(task.id) : Effect.void), ); if (deleted) { changed = true; From 278f960ddebf12d90a4fd11168670a07002703d1 Mon Sep 17 00:00:00 2001 From: JJ Eaton <58495321+oldbettie@users.noreply.github.com> Date: Fri, 13 Nov 2026 13:07:12 +0700 Subject: [PATCH 3/4] fix(server): stale-task sweep no longer turns away a restored webhook's delivery Cleanup took the run reservation while deleting, so a webhook restored and triggered in that window was refused as already running and its accepted delivery was recorded as dispatch_failed. Cleanup now only skips tasks that are running; the updated_at and status guards on the delete already drop it when a run, restore or edit lands first. Co-Authored-By: Claude Opus 5.5 --- .../scheduledTasks/ScheduledTaskService.ts | 41 +++++++--------- .../ScheduledTaskService.webhook.test.ts | 47 +++++++++++++++++++ 2 files changed, 65 insertions(+), 23 deletions(-) diff --git a/apps/server/src/scheduledTasks/ScheduledTaskService.ts b/apps/server/src/scheduledTasks/ScheduledTaskService.ts index 32871928a6617..5706e2adafc8a 100644 --- a/apps/server/src/scheduledTasks/ScheduledTaskService.ts +++ b/apps/server/src/scheduledTasks/ScheduledTaskService.ts @@ -2821,29 +2821,24 @@ export const layer = Layer.effect( }); } } else if (decision._tag === "delete") { - // Holding the run reservation keeps a run from starting mid-delete; the updated_at - // and status guards drop the decision if a run or edit landed since the read. - const deleted = yield* Effect.acquireUseRelease( - reserveRun(task.id), - (reserved) => - reserved - ? sql.withTransaction( - Effect.gen(function* () { - const removed = yield* sql` - DELETE FROM scheduled_tasks - WHERE task_id = ${row.task_id} AND archived_at = ${row.archived_at} - AND updated_at = ${row.updated_at} AND last_run_status <> 'running' - RETURNING task_id - `; - if (removed.length > 0) { - yield* sql`DELETE FROM scheduled_task_webhook_deliveries WHERE task_id = ${row.task_id}`; - } - return removed.length > 0; - }), - ) - : Effect.succeed(false), - (reserved) => (reserved ? releaseRun(task.id) : Effect.void), - ); + // The updated_at and status guards drop the decision if a run, restore or edit landed + // since the read. Cleanup never reserves the task, so it cannot turn away a run. + const deleted = (yield* Ref.get(activeRuns)).has(task.id) + ? false + : yield* sql.withTransaction( + Effect.gen(function* () { + const removed = yield* sql` + DELETE FROM scheduled_tasks + WHERE task_id = ${row.task_id} AND archived_at = ${row.archived_at} + AND updated_at = ${row.updated_at} AND last_run_status <> 'running' + RETURNING task_id + `; + if (removed.length > 0) { + yield* sql`DELETE FROM scheduled_task_webhook_deliveries WHERE task_id = ${row.task_id}`; + } + return removed.length > 0; + }), + ); if (deleted) { changed = true; yield* forgetWebhookTask(task.id); diff --git a/apps/server/src/scheduledTasks/ScheduledTaskService.webhook.test.ts b/apps/server/src/scheduledTasks/ScheduledTaskService.webhook.test.ts index da8685f55a4b5..8b3f1c9592ef1 100644 --- a/apps/server/src/scheduledTasks/ScheduledTaskService.webhook.test.ts +++ b/apps/server/src/scheduledTasks/ScheduledTaskService.webhook.test.ts @@ -2288,3 +2288,50 @@ it.effect("deletes an archived task and its deliveries 30 days after archiving i { workers: new Map() }, ), ); + +it.effect("a webhook restored while the sweep is about to delete it still runs its delivery", () => + Effect.gen(function* () { + const sql = yield* SqlClient.SqlClient; + const sweepPaused = yield* Deferred.make(); + const resumeSweep = yield* Deferred.make(); + let pauseNextTransaction = false; + // The delete is the sweep's only transaction, so this pauses it right before deleting. + const pausingSql = new Proxy(sql, { + get(target, key, receiver) { + if (key !== "withTransaction") return Reflect.get(target, key, receiver); + return (work: Effect.Effect) => + Effect.gen(function* () { + if (pauseNextTransaction) { + pauseNextTransaction = false; + yield* Deferred.succeed(sweepPaused, undefined); + yield* Deferred.await(resumeSweep); + } + return yield* sql.withTransaction(work); + }); + }, + }); + yield* withServiceOn( + pausingSql, + ({ service, launches }) => + Effect.gen(function* () { + const { task } = yield* service.upsert(yield* webhookTaskInput({ enabled: false })); + yield* TestClock.adjust(30 * CLEANUP_DAY_MS); + yield* service.cleanupStale(); + assert.equal((yield* service.list()).tasks[0]?.archiveReason, "webhook_unused"); + yield* TestClock.adjust(30 * CLEANUP_DAY_MS); + pauseNextTransaction = true; + const sweep = yield* service.cleanupStale().pipe(Effect.forkChild); + yield* Deferred.await(sweepPaused); + + yield* service.setEnabled({ id: task.id, enabled: true }); + assert.equal((yield* service.triggerWebhook(requestFor(task)))._tag, "accepted"); + yield* service.awaitWebhookDeliveries; + assert.equal(yield* Queue.size(launches), 1); + yield* Deferred.succeed(resumeSweep, undefined); + yield* Fiber.join(sweep); + assert.equal((yield* service.list()).tasks.length, 1); + }), + { workers: new Map() }, + ); + }).pipe(Effect.provide(SqlitePersistence.layerMemory)), +); From 3be595a8eda319135f6cfbade367e09f04a49fdb Mon Sep 17 00:00:00 2001 From: JJ Eaton <58495321+oldbettie@users.noreply.github.com> Date: Fri, 13 Nov 2026 13:25:51 +0700 Subject: [PATCH 4/4] fix(server): a run no longer starts on a task recreated under its id A run reads the task, then marks it running by id alone. If the task was deleted and recreated under the same id in between, as cleanup plus an idempotent create replay can do, the old run started on the replacement and stamped its history. The start now also matches the created_at it read and aborts when the row was replaced. Co-Authored-By: Claude Opus 5.5 --- .../ScheduledTaskService.test.ts | 51 +++++++++++++++++++ .../scheduledTasks/ScheduledTaskService.ts | 19 ++++--- 2 files changed, 64 insertions(+), 6 deletions(-) diff --git a/apps/server/src/scheduledTasks/ScheduledTaskService.test.ts b/apps/server/src/scheduledTasks/ScheduledTaskService.test.ts index 17d945aad2e44..576eec0b43b4e 100644 --- a/apps/server/src/scheduledTasks/ScheduledTaskService.test.ts +++ b/apps/server/src/scheduledTasks/ScheduledTaskService.test.ts @@ -1175,3 +1175,54 @@ it.effect("the sweep keeps an archived task whose run starts after the sweep rea }).pipe(Effect.provide(layer)); }).pipe(Effect.provide(SqlitePersistence.layerMemory)), ); + +it.effect( + "a run does not start on a task deleted and recreated under its id after it read it", + () => + Effect.gen(function* () { + yield* TestClock.setTime(CLEANUP_T0); + const sql = yield* SqlClient.SqlClient; + const startPaused = yield* Deferred.make(); + const resumeStart = yield* Deferred.make(); + let launches = 0; + // Pauses the run's start UPDATE, after it has read the task. + const pausingSql = new Proxy(sql, { + apply(target, thisArg, args: [TemplateStringsArray, ...Array]) { + const statement: ReturnType = Reflect.apply( + target, + thisArg, + args, + ); + return args[0].join("").includes("last_run_webhook_delivery_id = ") + ? Deferred.succeed(startPaused, undefined).pipe( + Effect.andThen(Deferred.await(resumeStart)), + Effect.andThen(statement), + ) + : statement; + }, + }); + const layer = cleanupLayer({ + launch: () => + Effect.sync(() => { + launches += 1; + return { threadId: ThreadId.make("thread:old-run"), resumed: false } as never; + }), + }); + yield* Effect.gen(function* () { + const service = yield* ScheduledTaskService.ScheduledTaskService; + const id = ScheduledTaskId.make("task:recreated"); + yield* service.upsert(cleanupInput(id, { enabled: false })); + const run = yield* service.runNow({ id }).pipe(Effect.exit, Effect.forkChild); + yield* Deferred.await(startPaused); + yield* service.delete({ id }); + yield* TestClock.adjust(1); + yield* service.upsert(cleanupInput(id, { enabled: false })); + yield* Deferred.succeed(resumeStart, undefined); + assert.isTrue(Exit.isFailure(yield* Fiber.join(run))); + assert.equal(launches, 0); + const [replacement] = (yield* service.list()).tasks; + assert.equal(replacement?.lastRunStatus, "never"); + assert.equal(replacement?.runCount, 0); + }).pipe(Effect.provide(layer), Effect.provideService(SqlClient.SqlClient, pausingSql)); + }).pipe(Effect.provide(SqlitePersistence.layerMemory)), +); diff --git a/apps/server/src/scheduledTasks/ScheduledTaskService.ts b/apps/server/src/scheduledTasks/ScheduledTaskService.ts index 5706e2adafc8a..cee5fb24ded33 100644 --- a/apps/server/src/scheduledTasks/ScheduledTaskService.ts +++ b/apps/server/src/scheduledTasks/ScheduledTaskService.ts @@ -818,7 +818,12 @@ export const layer = Layer.effect( ), ); - const markRunning = (id: ScheduledTaskId, startedAtIso: string, webhookDeliveryId?: string) => + // Matching created_at stops a run from starting on a task deleted and recreated under its id. + const markRunning = ( + task: Pick, + startedAtIso: string, + webhookDeliveryId?: string, + ) => sql<{ task_id: string }>` UPDATE scheduled_tasks SET updated_at = ${startedAtIso}, @@ -826,14 +831,16 @@ export const layer = Layer.effect( last_run_webhook_delivery_id = ${webhookDeliveryId ?? null}, last_run_status = 'running', last_run_error = NULL - WHERE task_id = ${id} + WHERE task_id = ${task.id} AND created_at = ${task.createdAt} RETURNING task_id `.pipe( Effect.mapError((cause) => - taskError("Could not mark schedule task as running.", { taskId: id, cause }), + taskError("Could not mark schedule task as running.", { taskId: task.id, cause }), ), Effect.flatMap((rows) => - rows.length > 0 ? Effect.void : taskError("Schedule task not found.", { taskId: id }), + rows.length > 0 + ? Effect.void + : taskError("Schedule task not found.", { taskId: task.id }), ), ); @@ -1118,12 +1125,12 @@ export const layer = Layer.effect( if (reason !== null) return yield* new WebhookDeliverySkipped({ reason }); } - if (webhook === undefined) yield* markRunning(active.id, startedAtIso); + if (webhook === undefined) yield* markRunning(active, startedAtIso); else yield* sql .withTransaction( Effect.gen(function* () { - yield* markRunning(active.id, startedAtIso, webhook.deliveryId); + yield* markRunning(active, startedAtIso, webhook.deliveryId); // Recovery must distinguish this delivery from other windows waiting on the same task. yield* sql`UPDATE scheduled_task_webhook_pending SET state = 'running' WHERE delivery_id = ${webhook.deliveryId}`; }),