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..576eec0b43b4e 100644 --- a/apps/server/src/scheduledTasks/ScheduledTaskService.test.ts +++ b/apps/server/src/scheduledTasks/ScheduledTaskService.test.ts @@ -926,3 +926,303 @@ 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)), +); + +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)), +); + +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 1efe55e3187f7..cee5fb24ded33 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?) => @@ -515,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 @@ -555,7 +578,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 +625,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 +781,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) => @@ -776,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}, @@ -784,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 }), ), ); @@ -811,7 +860,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 +898,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' @@ -1069,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}`; }), @@ -1184,26 +1240,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), ); }); @@ -1306,6 +1350,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 +1370,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 +1681,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 +1703,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 +2774,105 @@ 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") { + // 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); + 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 +2887,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..8b3f1c9592ef1 100644 --- a/apps/server/src/scheduledTasks/ScheduledTaskService.webhook.test.ts +++ b/apps/server/src/scheduledTasks/ScheduledTaskService.webhook.test.ts @@ -2230,3 +2230,108 @@ 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() }, + ), +); + +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)), +); 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), });