import { CommandId, MessageId, ScheduledTask, ScheduledTaskError, ScheduledTaskId, ThreadId, type ScheduledTaskDeleteInput, type ScheduledTaskDeleteResult, type ScheduledTaskListResult, type ScheduledTaskMutationResult, type ScheduledTaskRunNowInput, type ScheduledTaskRunNowResult, type ScheduledTaskSetEnabledInput, type ScheduledTaskUpsertInput,} from "@t3tools/contracts";import * as Cause from "effect/Cause";import * as Context from "effect/Context";import * as Crypto from "effect/Crypto";import * as DateTime from "effect/DateTime";import * as Duration from "effect/Duration";import * as Effect from "effect/Effect";import * as Layer from "effect/Layer";import * as PubSub from "effect/PubSub";import * as Ref from "effect/Ref";import * as Result from "effect/Result";import * as Schema from "effect/Schema";import * as Stream from "effect/Stream";import * as SqlClient from "effect/unstable/sql/SqlClient"; import * as ThreadLaunchService from "../orchestration-v2/ThreadLaunchService.ts";import * as ThreadManagementService from "../orchestration-v2/ThreadManagementService.ts";import { isMissedFixedTimeRun, isSameSchedule, nextScheduledRunAt } from "./Schedule.ts"; const decodeTask = Schema.decodeUnknownEffect(ScheduledTask);const decodeScheduleJson = Schema.decodeUnknownEffect( Schema.fromJsonString(ScheduledTask.fields.schedule),);const decodeWorkspaceStrategyJson = Schema.decodeUnknownEffect( Schema.fromJsonString(ScheduledTask.fields.workspaceStrategy),);const decodeModelSelectionJson = Schema.decodeUnknownEffect( Schema.fromJsonString(ScheduledTask.fields.modelSelection),); interface ScheduledTaskRow { readonly task_id: string; readonly title: string; readonly prompt: string; readonly enabled: number; readonly schedule_json: string; readonly project_id: string; readonly thread_id: string | null; readonly workspace_strategy_json: string; readonly model_selection_json: string; readonly runtime_mode: string; readonly interaction_mode: string; readonly created_by: string; readonly creation_source: string; readonly created_at: string; readonly updated_at: string; readonly next_run_at: string | null; readonly last_run_at: string | null; readonly last_run_status: string; readonly last_run_error: string | null; readonly run_count: number;} export class ScheduledTaskService extends Context.Service< ScheduledTaskService, { readonly list: () => Effect.Effect<ScheduledTaskListResult, ScheduledTaskError>; /** Emits the full task list on subscribe and again after every change (CRUD, run transitions, reschedules). */ readonly subscribeList: () => Stream.Stream<ScheduledTaskListResult, ScheduledTaskError>; readonly upsert: ( input: ScheduledTaskUpsertInput, ) => Effect.Effect<ScheduledTaskMutationResult, ScheduledTaskError>; /** Partial update flipping only the enabled flag; never touches other fields. */ readonly setEnabled: ( input: ScheduledTaskSetEnabledInput, ) => Effect.Effect<ScheduledTaskMutationResult, ScheduledTaskError>; readonly delete: ( input: ScheduledTaskDeleteInput, ) => Effect.Effect<ScheduledTaskDeleteResult, ScheduledTaskError>; readonly runNow: ( input: ScheduledTaskRunNowInput, ) => Effect.Effect<ScheduledTaskRunNowResult, ScheduledTaskError>; }>()("t3/scheduledTasks/ScheduledTaskService") {} function taskError(message: string, input?: { taskId?: ScheduledTaskId; cause?: unknown }) { return new ScheduledTaskError({ message, ...(input?.taskId === undefined ? {} : { taskId: input.taskId }), ...(input?.cause === undefined ? {} : { cause: input.cause }), });} function automationPrompt(task: ScheduledTask): string { return `[Triggered by schedule task: ${task.title}]\n\n${task.prompt}`;} function iso(value: DateTime.DateTime): string { return DateTime.formatIso(DateTime.toUtc(value));} const localNow = DateTime.withCurrentZoneLocal(DateTime.nowInCurrentZone); function nextRunAt( task: Pick<ScheduledTask, "enabled" | "schedule">, from: DateTime.DateTime,): string | null { if (!task.enabled) return null; const next = nextScheduledRunAt(task.schedule, from); return next === null ? null : iso(next);} function errorMessage(error: unknown): string { if (Cause.isCause(error)) return Cause.pretty(error); if (error instanceof Error) return error.message; return String(error);} const decodeRow = (row: ScheduledTaskRow) => Effect.gen(function* () { const id = ScheduledTaskId.make(row.task_id); const schedule = yield* decodeScheduleJson(row.schedule_json); const workspaceStrategy = yield* decodeWorkspaceStrategyJson(row.workspace_strategy_json); const modelSelection = yield* decodeModelSelectionJson(row.model_selection_json); return yield* decodeTask({ id, title: row.title, prompt: row.prompt, enabled: row.enabled === 1, schedule, projectId: row.project_id, threadId: row.thread_id, workspaceStrategy, modelSelection, runtimeMode: row.runtime_mode, interactionMode: row.interaction_mode, createdBy: row.created_by, creationSource: row.creation_source, createdAt: row.created_at, updatedAt: row.updated_at, nextRunAt: row.next_run_at, lastRunAt: row.last_run_at, lastRunStatus: row.last_run_status, lastRunError: row.last_run_error, runCount: row.run_count, }); }).pipe( Effect.mapError((cause) => taskError("Could not decode schedule task row.", { taskId: ScheduledTaskId.make(row.task_id), cause, }), ), ); export const layer = Layer.effect( ScheduledTaskService, Effect.gen(function* () { const sql = yield* SqlClient.SqlClient; const crypto = yield* Crypto.Crypto; const threadLaunch = yield* ThreadLaunchService.ThreadLaunchService; const threadManagement = yield* ThreadManagementService.ThreadManagementService; const activeRuns = yield* Ref.make<ReadonlySet<ScheduledTaskId>>(new Set()); // Sliding(1) coalesces the dirty-signal: every notification triggers a // full list() re-emit anyway, so a slow subscriber only ever needs the // latest signal — an unbounded backlog would just grow memory. const changesPubSub = yield* PubSub.sliding<void>(1); const notifyChanged = PubSub.publish(changesPubSub, undefined).pipe(Effect.asVoid); const selectAllRows = () => sql<ScheduledTaskRow>` SELECT task_id, title, prompt, enabled, schedule_json, project_id, thread_id, workspace_strategy_json, model_selection_json, runtime_mode, interaction_mode, created_by, creation_source, created_at, updated_at, next_run_at, last_run_at, last_run_status, last_run_error, run_count FROM scheduled_tasks ORDER BY updated_at DESC, task_id ASC `; // Strict decode for the API surface: a corrupt row is a visible error. const listRows = Effect.fn("ScheduledTaskService.listRows")(function* () { const rows = yield* selectAllRows(); return yield* Effect.forEach(rows, decodeRow, { concurrency: 1 }); }); // Lenient decode for the scheduler: one corrupt row must never halt the // poll loop or crash recovery for every other task — skip it and log. const listTasksLenient = Effect.fn("ScheduledTaskService.listTasksLenient")(function* () { const rows = yield* selectAllRows(); const tasks: ScheduledTask[] = []; for (const row of rows) { const decoded = yield* Effect.result(decodeRow(row)); if (Result.isSuccess(decoded)) { tasks.push(decoded.success); } else { yield* Effect.logWarning("Skipping undecodable schedule task row", { taskId: row.task_id, cause: decoded.failure, }); } } return tasks; }); const getRows = (id: ScheduledTaskId) => sql<ScheduledTaskRow>` SELECT task_id, title, prompt, enabled, schedule_json, project_id, thread_id, workspace_strategy_json, model_selection_json, runtime_mode, interaction_mode, created_by, creation_source, created_at, updated_at, next_run_at, last_run_at, last_run_status, last_run_error, run_count FROM scheduled_tasks WHERE task_id = ${id} `; /** Load a task, returning `null` when it does not exist; real load/decode failures propagate. */ const findTask = Effect.fn("ScheduledTaskService.findTask")(function* (id: ScheduledTaskId) { const rows = yield* getRows(id).pipe( Effect.mapError((cause) => taskError("Could not load schedule task.", { taskId: id, cause }), ), ); const row = rows[0]; if (row === undefined) return null; return yield* decodeRow(row); }); const loadTask = Effect.fn("ScheduledTaskService.loadTask")(function* (id: ScheduledTaskId) { const task = yield* findTask(id); if (task === null) { return yield* taskError("Schedule task not found.", { taskId: id }); } return task; }); // Run-state columns (last_run_*, run_count) are intentionally absent from // the conflict clause: they are owned by the run transitions below, and a // concurrent settings save must not overwrite an in-flight increment. const saveTask = (task: ScheduledTask) => sql` INSERT INTO scheduled_tasks ( task_id, title, prompt, enabled, schedule_json, project_id, thread_id, workspace_strategy_json, model_selection_json, runtime_mode, interaction_mode, created_by, creation_source, created_at, updated_at, next_run_at, last_run_at, last_run_status, last_run_error, run_count ) VALUES ( ${task.id}, ${task.title}, ${task.prompt}, ${task.enabled ? 1 : 0}, ${JSON.stringify(task.schedule)}, ${task.projectId}, ${task.threadId}, ${JSON.stringify(task.workspaceStrategy)}, ${JSON.stringify(task.modelSelection)}, ${task.runtimeMode}, ${task.interactionMode}, ${task.createdBy}, ${task.creationSource}, ${task.createdAt}, ${task.updatedAt}, ${task.nextRunAt}, ${task.lastRunAt}, ${task.lastRunStatus}, ${task.lastRunError}, ${task.runCount} ) ON CONFLICT (task_id) DO UPDATE SET title = excluded.title, prompt = excluded.prompt, enabled = excluded.enabled, schedule_json = excluded.schedule_json, project_id = excluded.project_id, thread_id = excluded.thread_id, workspace_strategy_json = excluded.workspace_strategy_json, model_selection_json = excluded.model_selection_json, runtime_mode = excluded.runtime_mode, interaction_mode = excluded.interaction_mode, creation_source = excluded.creation_source, updated_at = excluded.updated_at, next_run_at = excluded.next_run_at `.pipe( Effect.mapError((cause) => taskError("Could not save schedule task.", { taskId: task.id, cause }), ), ); const deleteRow = (id: ScheduledTaskId) => sql`DELETE FROM scheduled_tasks WHERE task_id = ${id}`.pipe( Effect.mapError((cause) => taskError("Could not delete schedule task.", { taskId: id, cause }), ), ); // Run-state transitions use targeted UPDATEs (never the full-row upsert) so // a completing run cannot resurrect a deleted task or clobber concurrent // edits to the task definition. const markRunning = (id: ScheduledTaskId, startedAtIso: string) => sql` UPDATE scheduled_tasks SET updated_at = ${startedAtIso}, last_run_at = ${startedAtIso}, last_run_status = 'running', last_run_error = NULL WHERE task_id = ${id} `.pipe( Effect.mapError((cause) => taskError("Could not mark schedule task as running.", { taskId: id, cause }), ), ); const markCompleted = (input: { readonly id: ScheduledTaskId; readonly completedAtIso: string; readonly nextRunAtIso: string | null; readonly status: "succeeded" | "failed"; readonly error: string | null; readonly startedAtIso: string; }) => sql` UPDATE scheduled_tasks SET updated_at = ${input.completedAtIso}, next_run_at = ${input.nextRunAtIso}, last_run_status = ${input.status}, last_run_error = ${input.error}, run_count = run_count + 1 WHERE task_id = ${input.id} AND last_run_status = 'running' AND last_run_at = ${input.startedAtIso} `.pipe( Effect.mapError((cause) => taskError("Could not record schedule task run.", { taskId: input.id, cause }), ), ); // Best-effort escape hatch: if anything fails between markRunning and // markCompleted, write a full terminal record so runDueTasks neither skips // the task forever (it filters out 'running' rows) nor re-fires it // immediately: the dispatch may already have gone out, so next_run_at must // advance and run_count must count the attempt. const releaseStuckRun = (task: ScheduledTask, message: string) => Effect.gen(function* () { const now = yield* localNow; // Compute the next occurrence from the current row so a schedule // edited while the run was in flight is honoured; fall back to the // run's snapshot only if the re-read itself fails. const reread = yield* Effect.result(findTask(task.id)); if (Result.isSuccess(reread) && reread.success === null) return; // deleted — nothing to release const source = Result.isSuccess(reread) && reread.success !== null ? reread.success : task; yield* sql` UPDATE scheduled_tasks SET last_run_status = 'failed', last_run_error = ${message}, next_run_at = ${nextRunAt(source, now)}, updated_at = ${iso(now)}, run_count = run_count + 1 WHERE task_id = ${task.id} AND last_run_status = 'running' `; yield* notifyChanged; }).pipe( Effect.catch((cause) => Effect.logWarning("Could not release stuck schedule task run", { taskId: task.id, cause, }), ), ); const runTask = Effect.fn("ScheduledTaskService.runTask")(function* ( task: ScheduledTask, trigger: "scheduled" | "manual", ) { const reserved = yield* 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; }); if (!reserved) { if (trigger === "manual") { return yield* taskError("Schedule task is already running.", { taskId: task.id }); } return task; } return yield* Effect.gen(function* () { const startedAt = yield* localNow; const startedAtIso = iso(startedAt); // The in-memory snapshot may be stale: re-read before touching run // state. The task may have been deleted, paused, or postponed since // the poll loaded it — none of those may fire. const active = yield* findTask(task.id); if (active === null) { // A manual run on a just-deleted task must fail loudly, not report // a successful run that never dispatched. if (trigger === "manual") { return yield* taskError("Schedule task not found.", { taskId: task.id }); } return task; } if ( trigger === "scheduled" && (!active.enabled || active.nextRunAt === null || DateTime.toEpochMillis(DateTime.makeUnsafe(active.nextRunAt)) > DateTime.toEpochMillis(startedAt)) ) { return active; } yield* markRunning(active.id, startedAtIso); yield* notifyChanged; const fireKey = `${active.id}:${DateTime.toEpochMillis(startedAt)}:${trigger}`; const commandId = CommandId.make(`scheduled-task:${fireKey}`); const messageId = MessageId.make(`scheduled-task-message:${fireKey}`); // Dispatch from the fresh row so prompt/model/binding edits made // after the poll read are honoured. const prompt = automationPrompt(active); // Effect.exit (not Effect.result) so defects and interruptions in the // dispatch are also captured and recorded as a failed run instead of // aborting before markCompleted. const result = active.threadId === null ? yield* Effect.exit( threadLaunch.launch({ commandId, projectId: active.projectId, title: active.title, modelSelection: active.modelSelection, runtimeMode: active.runtimeMode, interactionMode: active.interactionMode, workspaceStrategy: active.workspaceStrategy, initialMessage: { messageId, text: prompt, attachments: [], }, createdBy: active.createdBy, creationSource: active.creationSource, }), ) : yield* Effect.exit( threadManagement.sendToThread({Lines 1–500 of 799. Load more
import { CommandId, MessageId, ScheduledTask, ScheduledTaskError, ScheduledTaskId, ThreadId, type ScheduledTaskDeleteInput, type ScheduledTaskDeleteResult, type ScheduledTaskListResult, type ScheduledTaskMutationResult, type ScheduledTaskRunNowInput, type ScheduledTaskRunNowResult, type ScheduledTaskSetEnabledInput, type ScheduledTaskUpsertInput, } from "@t3tools/contracts"; import * as Cause from "effect/Cause"; import * as Context from "effect/Context"; import * as Crypto from "effect/Crypto"; import * as DateTime from "effect/DateTime"; import * as Duration from "effect/Duration"; import * as Effect from "effect/Effect"; import * as Layer from "effect/Layer"; import * as PubSub from "effect/PubSub"; import * as Ref from "effect/Ref"; import * as Result from "effect/Result"; import * as Schema from "effect/Schema"; import * as Stream from "effect/Stream"; import * as SqlClient from "effect/unstable/sql/SqlClient"; import * as ThreadLaunchService from "../orchestration-v2/ThreadLaunchService.ts"; import * as ThreadManagementService from "../orchestration-v2/ThreadManagementService.ts"; import { isMissedFixedTimeRun, isSameSchedule, nextScheduledRunAt } from "./Schedule.ts"; const decodeTask = Schema.decodeUnknownEffect(ScheduledTask); const decodeScheduleJson = Schema.decodeUnknownEffect( Schema.fromJsonString(ScheduledTask.fields.schedule), ); const decodeWorkspaceStrategyJson = Schema.decodeUnknownEffect( Schema.fromJsonString(ScheduledTask.fields.workspaceStrategy), ); const decodeModelSelectionJson = Schema.decodeUnknownEffect( Schema.fromJsonString(ScheduledTask.fields.modelSelection), ); interface ScheduledTaskRow { readonly task_id: string; readonly title: string; readonly prompt: string; readonly enabled: number; readonly schedule_json: string; readonly project_id: string; readonly thread_id: string | null; readonly workspace_strategy_json: string; readonly model_selection_json: string; readonly runtime_mode: string; readonly interaction_mode: string; readonly created_by: string; readonly creation_source: string; readonly created_at: string; readonly updated_at: string; readonly next_run_at: string | null; readonly last_run_at: string | null; readonly last_run_status: string; readonly last_run_error: string | null; readonly run_count: number; } export class ScheduledTaskService extends Context.Service< ScheduledTaskService, { readonly list: () => Effect.Effect<ScheduledTaskListResult, ScheduledTaskError>; /** Emits the full task list on subscribe and again after every change (CRUD, run transitions, reschedules). */ readonly subscribeList: () => Stream.Stream<ScheduledTaskListResult, ScheduledTaskError>; readonly upsert: ( input: ScheduledTaskUpsertInput, ) => Effect.Effect<ScheduledTaskMutationResult, ScheduledTaskError>; /** Partial update flipping only the enabled flag; never touches other fields. */ readonly setEnabled: ( input: ScheduledTaskSetEnabledInput, ) => Effect.Effect<ScheduledTaskMutationResult, ScheduledTaskError>; readonly delete: ( input: ScheduledTaskDeleteInput, ) => Effect.Effect<ScheduledTaskDeleteResult, ScheduledTaskError>; readonly runNow: ( input: ScheduledTaskRunNowInput, ) => Effect.Effect<ScheduledTaskRunNowResult, ScheduledTaskError>; } >()("t3/scheduledTasks/ScheduledTaskService") {} function taskError(message: string, input?: { taskId?: ScheduledTaskId; cause?: unknown }) { return new ScheduledTaskError({ message, ...(input?.taskId === undefined ? {} : { taskId: input.taskId }), ...(input?.cause === undefined ? {} : { cause: input.cause }), }); } function automationPrompt(task: ScheduledTask): string { return `[Triggered by schedule task: ${task.title}]\n\n${task.prompt}`; } function iso(value: DateTime.DateTime): string { return DateTime.formatIso(DateTime.toUtc(value)); } const localNow = DateTime.withCurrentZoneLocal(DateTime.nowInCurrentZone); function nextRunAt( task: Pick<ScheduledTask, "enabled" | "schedule">, from: DateTime.DateTime, ): string | null { if (!task.enabled) return null; const next = nextScheduledRunAt(task.schedule, from); return next === null ? null : iso(next); } function errorMessage(error: unknown): string { if (Cause.isCause(error)) return Cause.pretty(error); if (error instanceof Error) return error.message; return String(error); } const decodeRow = (row: ScheduledTaskRow) => Effect.gen(function* () { const id = ScheduledTaskId.make(row.task_id); const schedule = yield* decodeScheduleJson(row.schedule_json); const workspaceStrategy = yield* decodeWorkspaceStrategyJson(row.workspace_strategy_json); const modelSelection = yield* decodeModelSelectionJson(row.model_selection_json); return yield* decodeTask({ id, title: row.title, prompt: row.prompt, enabled: row.enabled === 1, schedule, projectId: row.project_id, threadId: row.thread_id, workspaceStrategy, modelSelection, runtimeMode: row.runtime_mode, interactionMode: row.interaction_mode, createdBy: row.created_by, creationSource: row.creation_source, createdAt: row.created_at, updatedAt: row.updated_at, nextRunAt: row.next_run_at, lastRunAt: row.last_run_at, lastRunStatus: row.last_run_status, lastRunError: row.last_run_error, runCount: row.run_count, }); }).pipe( Effect.mapError((cause) => taskError("Could not decode schedule task row.", { taskId: ScheduledTaskId.make(row.task_id), cause, }), ), ); export const layer = Layer.effect( ScheduledTaskService, Effect.gen(function* () { const sql = yield* SqlClient.SqlClient; const crypto = yield* Crypto.Crypto; const threadLaunch = yield* ThreadLaunchService.ThreadLaunchService; const threadManagement = yield* ThreadManagementService.ThreadManagementService; const activeRuns = yield* Ref.make<ReadonlySet<ScheduledTaskId>>(new Set()); // Sliding(1) coalesces the dirty-signal: every notification triggers a // full list() re-emit anyway, so a slow subscriber only ever needs the // latest signal — an unbounded backlog would just grow memory. const changesPubSub = yield* PubSub.sliding<void>(1); const notifyChanged = PubSub.publish(changesPubSub, undefined).pipe(Effect.asVoid); const selectAllRows = () => sql<ScheduledTaskRow>` SELECT task_id, title, prompt, enabled, schedule_json, project_id, thread_id, workspace_strategy_json, model_selection_json, runtime_mode, interaction_mode, created_by, creation_source, created_at, updated_at, next_run_at, last_run_at, last_run_status, last_run_error, run_count FROM scheduled_tasks ORDER BY updated_at DESC, task_id ASC `; // Strict decode for the API surface: a corrupt row is a visible error. const listRows = Effect.fn("ScheduledTaskService.listRows")(function* () { const rows = yield* selectAllRows(); return yield* Effect.forEach(rows, decodeRow, { concurrency: 1 }); }); // Lenient decode for the scheduler: one corrupt row must never halt the // poll loop or crash recovery for every other task — skip it and log. const listTasksLenient = Effect.fn("ScheduledTaskService.listTasksLenient")(function* () { const rows = yield* selectAllRows(); const tasks: ScheduledTask[] = []; for (const row of rows) { const decoded = yield* Effect.result(decodeRow(row)); if (Result.isSuccess(decoded)) { tasks.push(decoded.success); } else { yield* Effect.logWarning("Skipping undecodable schedule task row", { taskId: row.task_id, cause: decoded.failure, }); } } return tasks; }); const getRows = (id: ScheduledTaskId) => sql<ScheduledTaskRow>` SELECT task_id, title, prompt, enabled, schedule_json, project_id, thread_id, workspace_strategy_json, model_selection_json, runtime_mode, interaction_mode, created_by, creation_source, created_at, updated_at, next_run_at, last_run_at, last_run_status, last_run_error, run_count FROM scheduled_tasks WHERE task_id = ${id} `; /** Load a task, returning `null` when it does not exist; real load/decode failures propagate. */ const findTask = Effect.fn("ScheduledTaskService.findTask")(function* (id: ScheduledTaskId) { const rows = yield* getRows(id).pipe( Effect.mapError((cause) => taskError("Could not load schedule task.", { taskId: id, cause }), ), ); const row = rows[0]; if (row === undefined) return null; return yield* decodeRow(row); }); const loadTask = Effect.fn("ScheduledTaskService.loadTask")(function* (id: ScheduledTaskId) { const task = yield* findTask(id); if (task === null) { return yield* taskError("Schedule task not found.", { taskId: id }); } return task; }); // Run-state columns (last_run_*, run_count) are intentionally absent from // the conflict clause: they are owned by the run transitions below, and a // concurrent settings save must not overwrite an in-flight increment. const saveTask = (task: ScheduledTask) => sql` INSERT INTO scheduled_tasks ( task_id, title, prompt, enabled, schedule_json, project_id, thread_id, workspace_strategy_json, model_selection_json, runtime_mode, interaction_mode, created_by, creation_source, created_at, updated_at, next_run_at, last_run_at, last_run_status, last_run_error, run_count ) VALUES ( ${task.id}, ${task.title}, ${task.prompt}, ${task.enabled ? 1 : 0}, ${JSON.stringify(task.schedule)}, ${task.projectId}, ${task.threadId}, ${JSON.stringify(task.workspaceStrategy)}, ${JSON.stringify(task.modelSelection)}, ${task.runtimeMode}, ${task.interactionMode}, ${task.createdBy}, ${task.creationSource}, ${task.createdAt}, ${task.updatedAt}, ${task.nextRunAt}, ${task.lastRunAt}, ${task.lastRunStatus}, ${task.lastRunError}, ${task.runCount} ) ON CONFLICT (task_id) DO UPDATE SET title = excluded.title, prompt = excluded.prompt, enabled = excluded.enabled, schedule_json = excluded.schedule_json, project_id = excluded.project_id, thread_id = excluded.thread_id, workspace_strategy_json = excluded.workspace_strategy_json, model_selection_json = excluded.model_selection_json, runtime_mode = excluded.runtime_mode, interaction_mode = excluded.interaction_mode, creation_source = excluded.creation_source, updated_at = excluded.updated_at, next_run_at = excluded.next_run_at `.pipe( Effect.mapError((cause) => taskError("Could not save schedule task.", { taskId: task.id, cause }), ), ); const deleteRow = (id: ScheduledTaskId) => sql`DELETE FROM scheduled_tasks WHERE task_id = ${id}`.pipe( Effect.mapError((cause) => taskError("Could not delete schedule task.", { taskId: id, cause }), ), ); // Run-state transitions use targeted UPDATEs (never the full-row upsert) so // a completing run cannot resurrect a deleted task or clobber concurrent // edits to the task definition. const markRunning = (id: ScheduledTaskId, startedAtIso: string) => sql` UPDATE scheduled_tasks SET updated_at = ${startedAtIso}, last_run_at = ${startedAtIso}, last_run_status = 'running', last_run_error = NULL WHERE task_id = ${id} `.pipe( Effect.mapError((cause) => taskError("Could not mark schedule task as running.", { taskId: id, cause }), ), ); const markCompleted = (input: { readonly id: ScheduledTaskId; readonly completedAtIso: string; readonly nextRunAtIso: string | null; readonly status: "succeeded" | "failed"; readonly error: string | null; readonly startedAtIso: string; }) => sql` UPDATE scheduled_tasks SET updated_at = ${input.completedAtIso}, next_run_at = ${input.nextRunAtIso}, last_run_status = ${input.status}, last_run_error = ${input.error}, run_count = run_count + 1 WHERE task_id = ${input.id} AND last_run_status = 'running' AND last_run_at = ${input.startedAtIso} `.pipe( Effect.mapError((cause) => taskError("Could not record schedule task run.", { taskId: input.id, cause }), ), ); // Best-effort escape hatch: if anything fails between markRunning and // markCompleted, write a full terminal record so runDueTasks neither skips // the task forever (it filters out 'running' rows) nor re-fires it // immediately: the dispatch may already have gone out, so next_run_at must // advance and run_count must count the attempt. const releaseStuckRun = (task: ScheduledTask, message: string) => Effect.gen(function* () { const now = yield* localNow; // Compute the next occurrence from the current row so a schedule // edited while the run was in flight is honoured; fall back to the // run's snapshot only if the re-read itself fails. const reread = yield* Effect.result(findTask(task.id)); if (Result.isSuccess(reread) && reread.success === null) return; // deleted — nothing to release const source = Result.isSuccess(reread) && reread.success !== null ? reread.success : task; yield* sql` UPDATE scheduled_tasks SET last_run_status = 'failed', last_run_error = ${message}, next_run_at = ${nextRunAt(source, now)}, updated_at = ${iso(now)}, run_count = run_count + 1 WHERE task_id = ${task.id} AND last_run_status = 'running' `; yield* notifyChanged; }).pipe( Effect.catch((cause) => Effect.logWarning("Could not release stuck schedule task run", { taskId: task.id, cause, }), ), ); const runTask = Effect.fn("ScheduledTaskService.runTask")(function* ( task: ScheduledTask, trigger: "scheduled" | "manual", ) { const reserved = yield* 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; }); if (!reserved) { if (trigger === "manual") { return yield* taskError("Schedule task is already running.", { taskId: task.id }); } return task; } return yield* Effect.gen(function* () { const startedAt = yield* localNow; const startedAtIso = iso(startedAt); // The in-memory snapshot may be stale: re-read before touching run // state. The task may have been deleted, paused, or postponed since // the poll loaded it — none of those may fire. const active = yield* findTask(task.id); if (active === null) { // A manual run on a just-deleted task must fail loudly, not report // a successful run that never dispatched. if (trigger === "manual") { return yield* taskError("Schedule task not found.", { taskId: task.id }); } return task; } if ( trigger === "scheduled" && (!active.enabled || active.nextRunAt === null || DateTime.toEpochMillis(DateTime.makeUnsafe(active.nextRunAt)) > DateTime.toEpochMillis(startedAt)) ) { return active; } yield* markRunning(active.id, startedAtIso); yield* notifyChanged; const fireKey = `${active.id}:${DateTime.toEpochMillis(startedAt)}:${trigger}`; const commandId = CommandId.make(`scheduled-task:${fireKey}`); const messageId = MessageId.make(`scheduled-task-message:${fireKey}`); // Dispatch from the fresh row so prompt/model/binding edits made // after the poll read are honoured. const prompt = automationPrompt(active); // Effect.exit (not Effect.result) so defects and interruptions in the // dispatch are also captured and recorded as a failed run instead of // aborting before markCompleted. const result = active.threadId === null ? yield* Effect.exit( threadLaunch.launch({ commandId, projectId: active.projectId, title: active.title, modelSelection: active.modelSelection, runtimeMode: active.runtimeMode, interactionMode: active.interactionMode, workspaceStrategy: active.workspaceStrategy, initialMessage: { messageId, text: prompt, attachments: [], }, createdBy: active.createdBy, creationSource: active.creationSource, }), ) : yield* Effect.exit( threadManagement.sendToThread({