From 59ed76f3006b9cab0215f94aa7de8b42bdd8ef77 Mon Sep 17 00:00:00 2001 From: "github-actions[bot]" <41898282+github-actions[bot]@users.noreply.github.com> Date: Tue, 6 Oct 2026 22:23:31 +1100 Subject: [PATCH 1/8] fix(server): make scheduled-task edits atomic Squash of the six #11592 commits (70ba40b557..1d236a6fe8) so the rebase onto main's webhook-trigger scheduled tasks resolves once: - project-scoped partial `update` RPC that writes only the provided fields in one transaction and never inserts - upsert, enablement, dispatch, completion and recovery re-read the row in their write transaction, retrying SQLITE_BUSY_SNAPSHOT on a fresh snapshot - web and mobile editors send dirty-field patches (sparse workspaceStrategyPatch, mobile runtimeMode) when the server advertises scheduledTaskUpdate - thread binding validated only when the write changes it - MCP edits pinned to the modes that were authorized Co-Authored-By: Claude Opus 5.5 (1M context) --- .../SettingsScheduledTasksRouteScreen.tsx | 83 +- .../settings/scheduledTaskDraft.test.ts | 140 ++ .../features/settings/scheduledTaskDraft.ts | 146 +- .../src/environment/ServerEnvironment.test.ts | 1 + .../src/environment/ServerEnvironment.ts | 1 + apps/server/src/mcp/OrchestratorMcpService.ts | 93 +- ...OrchestratorMcpToolkit.integration.test.ts | 79 +- apps/server/src/mcp/toolkits/core.test.ts | 11 +- .../src/mcp/toolkits/thread/handlers.ts | 8 +- .../src/observability/RpcInstrumentation.ts | 1 + .../ScheduledTaskService.test.ts | 1215 ++++++++++++++++- .../scheduledTasks/ScheduledTaskService.ts | 891 +++++++++--- apps/server/src/ws.ts | 25 + .../settings/ScheduledTasksSettings.tsx | 105 +- .../settings/scheduledTasks.logic.test.ts | 587 ++++++++ .../settings/scheduledTasks.logic.ts | 201 +++ packages/client-runtime/src/state/server.ts | 6 + .../contracts/src/clientRpcPermissions.ts | 1 + packages/contracts/src/environment.test.ts | 9 + packages/contracts/src/environment.ts | 2 + packages/contracts/src/rpc.ts | 9 + packages/contracts/src/scheduledTask.ts | 85 ++ 22 files changed, 3385 insertions(+), 314 deletions(-) create mode 100644 apps/web/src/components/settings/scheduledTasks.logic.test.ts create mode 100644 apps/web/src/components/settings/scheduledTasks.logic.ts diff --git a/apps/mobile/src/features/settings/SettingsScheduledTasksRouteScreen.tsx b/apps/mobile/src/features/settings/SettingsScheduledTasksRouteScreen.tsx index e906b33020a7..4b16ba0cafd7 100644 --- a/apps/mobile/src/features/settings/SettingsScheduledTasksRouteScreen.tsx +++ b/apps/mobile/src/features/settings/SettingsScheduledTasksRouteScreen.tsx @@ -2,6 +2,7 @@ import { useAtomValue } from "@effect/atom-react"; import { AuthOrchestrationOperateScope, resolveProjectProviderInstanceEnabled, + scheduledTaskLegacyUpsert, } from "@t3tools/contracts"; import { readEnvironmentScope } from "../../state/session"; import type { @@ -71,6 +72,7 @@ import { SettingsScreen } from "./components/SettingsScreen"; import { SettingsSection } from "./components/SettingsSection"; import { useSettingsEnvironmentFilter, type SettingsTarget } from "./settings-environment-filter"; import { + buildScheduledTaskUpdateInput, editDraft, scheduledTaskDefaultModel, scheduleFromDraft, @@ -596,6 +598,10 @@ function TaskForm({ label: "scheduled task upsert", reportFailure: false, }); + const update = useAtomCommand(serverEnvironment.updateScheduledTask, { + label: "scheduled task update", + reportFailure: false, + }); const submissionPending = useRef(false); const [timePickerOpen, setTimePickerOpen] = useState(false); const taskMissing = @@ -658,35 +664,62 @@ function TaskForm({ Alert.alert("Project unavailable", "Choose a project in this environment."); return; } - const input: ScheduledTaskUpsertInput = { - ...(draft.task ? { id: draft.task.id, requireExisting: true } : {}), - title: draft.title.trim(), - prompt: draft.prompt.trim(), - projectId: draft.projectId, - modelSelection: draft.modelSelection, - schedule, - enabled: draft.enabled, - threadId: draft.task?.threadId ?? null, - workspaceStrategy: - draft.workspace === "root" - ? { type: "root" } - : draft.workspace === "existing_worktree" - ? { type: "existing_worktree", worktreePath: draft.checkoutPath.trim() } - : { - type: "worktree", - baseRef: draft.baseRef.trim() || "main", - startFromOrigin: draft.startFromOrigin, - }, - runtimeMode: draft.runtimeMode, - interactionMode: draft.task?.interactionMode ?? "default", - creationSource: draft.task?.creationSource ?? "mobile", - }; // Lock before React renders, and keep successful creates locked until the form closes. submissionPending.current = true; setSaving(true); - const result = await upsert({ environmentId, input }); + let result: AtomCommandResult | null; + if (draft.task !== null) { + // Existing tasks save as a dirty-field patch through the atomic update + // path: a stale editor can never overwrite fields another client + // changed, and a delete racing the save is a typed not-found, not a + // resurrection. `taskMissing` already guarantees the task is in the + // live list. + const liveTask = tasks.data?.tasks.find((task) => task.id === draft.task?.id); + if (liveTask === undefined) { + // The live list has not returned (or the task vanished between render + // and save) — fail loudly instead of silently discarding the edit. + setSaving(false); + submissionPending.current = false; + Alert.alert( + "Task unavailable", + "The task could not be loaded. Reopen it once the list has loaded and try again.", + ); + return; + } + const patch = buildScheduledTaskUpdateInput(draft, liveTask); + result = + patch === null + ? null + : config?.environment.capabilities.scheduledTaskUpdate === true + ? await update({ environmentId, input: patch }) + : await upsert({ environmentId, input: scheduledTaskLegacyUpsert(liveTask, patch) }); + } else { + const input: ScheduledTaskUpsertInput = { + title: draft.title.trim(), + prompt: draft.prompt.trim(), + projectId: draft.projectId, + modelSelection: draft.modelSelection, + schedule, + enabled: draft.enabled, + threadId: null, + workspaceStrategy: + draft.workspace === "root" + ? { type: "root" } + : draft.workspace === "existing_worktree" + ? { type: "existing_worktree", worktreePath: draft.checkoutPath.trim() } + : { + type: "worktree", + baseRef: draft.baseRef.trim() || "main", + startFromOrigin: draft.startFromOrigin, + }, + runtimeMode: draft.runtimeMode, + interactionMode: "default", + creationSource: "mobile", + }; + result = await upsert({ environmentId, input }); + } setSaving(false); - if (result._tag === "Failure") { + if (result !== null && result._tag === "Failure") { submissionPending.current = false; failure("Could not save task", result); return; diff --git a/apps/mobile/src/features/settings/scheduledTaskDraft.test.ts b/apps/mobile/src/features/settings/scheduledTaskDraft.test.ts index d69c5124d9b7..981d0b127794 100644 --- a/apps/mobile/src/features/settings/scheduledTaskDraft.test.ts +++ b/apps/mobile/src/features/settings/scheduledTaskDraft.test.ts @@ -9,6 +9,7 @@ import { type ScheduledTask, } from "@t3tools/contracts"; import { + buildScheduledTaskUpdateInput, scheduledTaskDefaultModel, createDraft, editDraft, @@ -353,3 +354,142 @@ describe("scheduled task model defaults", () => { ).toBeNull(); }); }); + +describe("buildScheduledTaskUpdateInput", () => { + const projectId = ProjectId.make("project:one"); + const task: ScheduledTask = { + id: ScheduledTaskId.make("scheduled-task:mobile-fixture"), + title: "Original title", + prompt: "Original prompt", + enabled: true, + schedule: { type: "interval", everyMs: 15 * 60_000 }, + projectId, + threadId: null, + workspaceStrategy: { type: "worktree", baseRef: "main", startFromOrigin: true }, + modelSelection: { + instanceId: ProviderInstanceId.make("codex"), + model: "gpt-5.4", + }, + runtimeMode: "full-access", + interactionMode: "default", + createdBy: "user", + creationSource: "mobile", + createdAt: "2026-09-01T00:00:00.000Z", + updatedAt: "2026-09-01T00:00:00.000Z", + nextRunAt: "2026-09-02T00:00:00.000Z", + lastRunAt: null, + lastRunStatus: "never", + lastRunError: null, + runCount: 0, + }; + + it("returns null when the draft matches the task it opened with", () => { + expect(buildScheduledTaskUpdateInput(editDraft(task), task)).toBeNull(); + expect(buildScheduledTaskUpdateInput(createDraft(null, null), task)).toBeNull(); + }); + + it("sends only the fields a stale editor changed, preserving concurrent edits", () => { + // Another client rewrote the prompt and paused the task while this + // editor was open; the mobile save carries only the rename the user + // actually made instead of restoring the stale prompt and enabled flag. + const draft = { ...editDraft(task), title: "Renamed" }; + const live: ScheduledTask = { + ...task, + prompt: "Rewritten elsewhere", + enabled: false, + }; + expect(buildScheduledTaskUpdateInput(draft, live)).toEqual({ + id: task.id, + projectId: task.projectId, + title: "Renamed", + }); + }); + + it("leaves an unchanged webhook trigger out of the patch", () => { + const webhookTask: ScheduledTask = { + ...task, + schedule: { + type: "webhook", + signature: { header: "x-hub-signature-256", encoding: "hex", prefix: "sha256=" }, + maxDeliveryAgeMinutes: 30, + }, + }; + const draft = { ...editDraft(webhookTask), title: "Renamed" }; + expect(buildScheduledTaskUpdateInput(draft, webhookTask)).toEqual({ + id: webhookTask.id, + projectId: webhookTask.projectId, + title: "Renamed", + }); + }); + + it("patches the editable runtime mode", () => { + const draft = { ...editDraft(task), runtimeMode: "auto" as const }; + expect(buildScheduledTaskUpdateInput(draft, task)).toEqual({ + id: task.id, + projectId: task.projectId, + runtimeMode: "auto", + }); + }); + + it("detaches the live binding only when the patch itself re-homes the task", () => { + const other = ProjectId.make("project:other"); + const bound: ScheduledTask = { ...task, threadId: "thread:bound" as ScheduledTask["threadId"] }; + // A real move away from the binding's project must detach it. + const moving = { ...editDraft(bound), projectId: other }; + expect(buildScheduledTaskUpdateInput(moving, bound)).toEqual({ + id: bound.id, + projectId: bound.projectId, + nextProjectId: other, + threadId: null, + }); + // A stale save landing where the task already lives keeps a concurrent + // rebind instead of erasing it. + const liveMoved: ScheduledTask = { ...bound, projectId: other }; + expect(buildScheduledTaskUpdateInput(moving, liveMoved)).toEqual({ + id: bound.id, + projectId: other, + nextProjectId: other, + }); + }); + + it("emits the normalized schedule for a legacy sub-minute interval", () => { + const legacy: ScheduledTask = { + ...task, + schedule: { type: "interval", everyMs: 30_000 }, + }; + // The draft is untouched — editDraft clamped the baseline to "1" — but + // the stored 30s interval must still be rewritten to the displayed 60s. + expect(buildScheduledTaskUpdateInput(editDraft(legacy), legacy)).toEqual({ + id: legacy.id, + projectId: legacy.projectId, + schedule: { type: "interval", everyMs: 60_000 }, + }); + }); + + it("does not overwrite a concurrent fix to a legacy sub-minute interval", () => { + const legacy: ScheduledTask = { + ...task, + schedule: { type: "interval", everyMs: 30_000 }, + }; + // Another client already normalized the schedule — an untouched save + // leaves their interval alone instead of forcing the editor's 60s. + const live: ScheduledTask = { ...legacy, schedule: { type: "interval", everyMs: 90_000 } }; + expect(buildScheduledTaskUpdateInput(editDraft(legacy), live)).toBeNull(); + }); + + it("sends only the workspace controls the user changed as a sparse patch", () => { + // Another client flipped startFromOrigin while this editor was open; + // the save sends just the edited key and the server merges it into the + // live strategy, so the concurrent toggle survives on both sides. + const draft = { ...editDraft(task), baseRef: "develop" }; + const live: ScheduledTask = { + ...task, + workspaceStrategy: { type: "worktree", baseRef: "main", startFromOrigin: false }, + }; + expect(buildScheduledTaskUpdateInput(draft, live)).toEqual({ + id: task.id, + projectId: task.projectId, + workspaceStrategyPatch: { baseRef: "develop" }, + }); + }); +}); diff --git a/apps/mobile/src/features/settings/scheduledTaskDraft.ts b/apps/mobile/src/features/settings/scheduledTaskDraft.ts index 707bd987e822..a989fd8e5d4a 100644 --- a/apps/mobile/src/features/settings/scheduledTaskDraft.ts +++ b/apps/mobile/src/features/settings/scheduledTaskDraft.ts @@ -1,14 +1,16 @@ import type { ModelSelection, + OrchestrationV2ThreadLaunchWorkspaceStrategy, ServerConfig, ProjectId, RuntimeMode, ScheduledTask, + ScheduledTaskUpdateInput, ScheduledTaskUpsertSchedule, ScheduledTaskWebhookSignature, } from "@t3tools/contracts"; -import { DEFAULT_SERVER_SETTINGS } from "@t3tools/contracts"; +import { DEFAULT_SERVER_SETTINGS, MIN_SCHEDULED_TASK_INTERVAL_MS } from "@t3tools/contracts"; import { parseMaxDeliveryAge } from "@t3tools/client-runtime/scheduled-task-webhook"; import { resolveProjectSettings, @@ -223,3 +225,145 @@ export function editDraft(task: ScheduledTask): ScheduledTaskDraft { runtimeMode: task.runtimeMode, }; } + +function sameSchedule(a: ScheduledTaskUpsertSchedule, b: ScheduledTaskUpsertSchedule): boolean { + if (a.type === "interval") { + return b.type === "interval" && a.everyMs === b.everyMs; + } + if (a.type === "webhook") { + if (b.type !== "webhook") return false; + const signatureKey = (signature: typeof a.signature) => + signature == null + ? null + : JSON.stringify([signature.header.toLowerCase(), signature.encoding, signature.prefix]); + return ( + signatureKey(a.signature) === signatureKey(b.signature) && + (a.maxDeliveryAgeMinutes ?? null) === (b.maxDeliveryAgeMinutes ?? null) + ); + } + if (b.type !== "fixed_time") return false; + const key = (weekdays: ReadonlyArray | undefined) => { + const unique = [...new Set(weekdays ?? [])].sort((x, y) => x - y); + return unique.length === 0 || unique.length === 7 ? "daily" : unique.join(","); + }; + return a.timeOfDay === b.timeOfDay && key(a.weekdays) === key(b.weekdays); +} + +function sameModelSelection(a: ModelSelection | null, b: ModelSelection | null): boolean { + if (a === null || b === null) return a === b; + const key = (selection: ModelSelection) => + JSON.stringify([ + selection.instanceId, + selection.model, + [...(selection.options ?? [])] + .sort((x, y) => x.id.localeCompare(y.id)) + .map((option) => [option.id, option.value]), + ]); + return key(a) === key(b); +} + +function workspaceStrategyFromDraft( + draft: ScheduledTaskDraft, +): OrchestrationV2ThreadLaunchWorkspaceStrategy { + if (draft.workspace === "root") return { type: "root" }; + if (draft.workspace === "existing_worktree") { + return { type: "existing_worktree", worktreePath: draft.checkoutPath.trim() }; + } + return { + type: "worktree", + baseRef: draft.baseRef.trim() || "main", + startFromOrigin: draft.startFromOrigin, + }; +} + +/** + * Dirty-field patch for an existing task, scoped to its live project. The + * baseline is the task snapshot the editor opened with (`draft.task`), so a + * concurrent commit landing mid-session is never read as a change the user + * made — and `projectId` tracks the live row, so a save still targets the + * task wherever it currently lives. Returns null when nothing changed. + */ +export function buildScheduledTaskUpdateInput( + draft: ScheduledTaskDraft, + liveTask: ScheduledTask, +): ScheduledTaskUpdateInput | null { + const opening = draft.task; + if (opening === null) return null; + const baseline = editDraft(opening); + const patch: { + -readonly [ + K in Exclude + ]?: ScheduledTaskUpdateInput[K]; + } = {}; + const title = draft.title.trim(); + if (title !== baseline.title.trim()) patch.title = title; + const prompt = draft.prompt.trim(); + if (prompt !== baseline.prompt.trim()) patch.prompt = prompt; + if (draft.enabled !== baseline.enabled) patch.enabled = draft.enabled; + const schedule = scheduleFromDraft(draft.schedule); + const baselineSchedule = scheduleFromDraft(baseline.schedule); + if ( + schedule !== null && + baselineSchedule !== null && + (!sameSchedule(schedule, baselineSchedule) || + // scheduleDraftForTask clamps a legacy sub-minute interval to one + // minute, so the baseline diff cannot see the normalization the + // editor promises. Emit the normalized schedule while the live task + // still carries it — a concurrent write to a valid interval clears + // this condition instead of being overwritten. + (liveTask.schedule.type === "interval" && + liveTask.schedule.everyMs < MIN_SCHEDULED_TASK_INTERVAL_MS)) + ) { + patch.schedule = schedule; + } + if (draft.runtimeMode !== baseline.runtimeMode) patch.runtimeMode = draft.runtimeMode; + if ( + draft.modelSelection !== null && + !sameModelSelection(draft.modelSelection, baseline.modelSelection) + ) { + patch.modelSelection = draft.modelSelection; + } + const workspaceStrategy = workspaceStrategyFromDraft(draft); + const baselineStrategy = workspaceStrategyFromDraft(baseline); + if (JSON.stringify(workspaceStrategy) !== JSON.stringify(baselineStrategy)) { + if (workspaceStrategy.type !== baselineStrategy.type) { + patch.workspaceStrategy = workspaceStrategy; + } else if ( + workspaceStrategy.type === "worktree" && + baselineStrategy.type === "worktree" && + liveTask.workspaceStrategy.type === "worktree" + ) { + // Send only the controls the user changed: the server merges the + // sparse patch into the live strategy inside the update transaction, + // so a concurrent edit to an untouched control — or to a field the + // editor cannot express (`branch`) — survives on both sides. + patch.workspaceStrategyPatch = { + ...(workspaceStrategy.baseRef !== baselineStrategy.baseRef + ? { baseRef: workspaceStrategy.baseRef } + : {}), + ...(workspaceStrategy.startFromOrigin !== baselineStrategy.startFromOrigin + ? { startFromOrigin: workspaceStrategy.startFromOrigin } + : {}), + }; + } else if ( + workspaceStrategy.type === "existing_worktree" && + baselineStrategy.type === "existing_worktree" && + liveTask.workspaceStrategy.type === "existing_worktree" + ) { + patch.workspaceStrategyPatch = { worktreePath: workspaceStrategy.worktreePath }; + } + // A live kind the draft no longer matches means a concurrent editor + // switched it — drop the stale control edit rather than revert that. + } + if (draft.projectId !== null && draft.projectId !== baseline.projectId) { + patch.nextProjectId = draft.projectId; + // A binding cannot follow a real re-home — the thread stays in the old + // project — so a patch that moves the task detaches it; a stale save + // landing where the task already lives keeps a concurrent rebind. + if (draft.projectId !== liveTask.projectId && liveTask.threadId !== null) { + patch.threadId = null; + } + } + if (Object.keys(patch).length === 0) return null; + return { id: liveTask.id, projectId: liveTask.projectId, ...patch }; +} diff --git a/apps/server/src/environment/ServerEnvironment.test.ts b/apps/server/src/environment/ServerEnvironment.test.ts index 1d2258b18ab1..18ba4c0bf2c7 100644 --- a/apps/server/src/environment/ServerEnvironment.test.ts +++ b/apps/server/src/environment/ServerEnvironment.test.ts @@ -211,6 +211,7 @@ it.layer(NodeServices.layer)("ServerEnvironmentLive", (it) => { expect(first.environmentId).toBe(second.environmentId); expect(first.orchestrationProtocolVersion).toBe(ORCHESTRATION_PROTOCOL_VERSION); expect(second.capabilities.repositoryIdentity).toBe(true); + expect(second.capabilities.scheduledTaskUpdate).toBe(true); expect(second.capabilities.connectionProbe).toBe(true); expect(second.capabilities.attachmentUploads).toBe(true); expect(second.capabilities.fileAttachments).toEqual({ maxUploadBytes: 50 * 1024 * 1024 }); diff --git a/apps/server/src/environment/ServerEnvironment.ts b/apps/server/src/environment/ServerEnvironment.ts index a55b1098d5f0..72b861380013 100644 --- a/apps/server/src/environment/ServerEnvironment.ts +++ b/apps/server/src/environment/ServerEnvironment.ts @@ -226,6 +226,7 @@ export const make = Effect.gen(function* () { inlineMessageContext: true, requiredWorktreeBootstrap: true, threadSettlement: true, + scheduledTaskUpdate: true, threadAutoSettlement: true, storageCleanup: true, storageCleanupRun: true, diff --git a/apps/server/src/mcp/OrchestratorMcpService.ts b/apps/server/src/mcp/OrchestratorMcpService.ts index f9bb61fa810e..9ba3258e318a 100644 --- a/apps/server/src/mcp/OrchestratorMcpService.ts +++ b/apps/server/src/mcp/OrchestratorMcpService.ts @@ -1705,56 +1705,75 @@ const make = Effect.gen(function* () { : "bindToCurrentThread binds to this thread, which belongs to a different project.", ); } - const threadId = - input.bindToCurrentThread === undefined - ? existing.threadId - : input.bindToCurrentThread && parent !== undefined - ? parent.thread.id - : null; - // Rebinding changes where runs execute, so the workspace strategy must - // follow: unbinding a root-strategy task would otherwise run loose - // prompts in the shared project checkout. - const workspaceStrategy = - input.bindToCurrentThread === undefined - ? existing.workspaceStrategy - : scheduledTaskWorkspaceStrategy(input.bindToCurrentThread); - const upsertInput: ScheduledTaskUpsertInput = { - id: existing.id, - title: input.title ?? existing.title, - prompt: input.prompt ?? existing.prompt, - enabled: input.enabled ?? existing.enabled, - schedule: input.schedule ?? existing.schedule, - projectId: existing.projectId, - threadId, - workspaceStrategy, - modelSelection: existing.modelSelection, - runtimeMode: existing.runtimeMode, - interactionMode: existing.interactionMode, - createdBy: existing.createdBy, - creationSource: existing.creationSource, - }; - const { task } = yield* scheduledTasks - .upsert(upsertInput) + // Atomic scoped partial update: only the fields the caller provided + // are written, and only while the task still exists in the project it + // was authorized in. A delete or move racing the edit surfaces as + // task_not_found instead of a stale full-row upsert resurrecting it. + const updated = yield* scheduledTasks + .update({ + id: existing.id, + projectId: existing.projectId, + // loadScheduledTask authorized these modes; a raise committed + // since must fail the edit rather than run the caller's patch + // above its limits. + expectedRuntimeMode: existing.runtimeMode, + expectedInteractionMode: existing.interactionMode, + ...(input.title === undefined ? {} : { title: input.title }), + ...(input.prompt === undefined ? {} : { prompt: input.prompt }), + ...(input.enabled === undefined ? {} : { enabled: input.enabled }), + ...(input.schedule === undefined ? {} : { schedule: input.schedule }), + // Rebinding changes where runs execute, so the workspace strategy + // must follow: unbinding a root-strategy task would otherwise run + // loose prompts in the shared project checkout. + ...(input.bindToCurrentThread === undefined + ? {} + : { + threadId: + input.bindToCurrentThread && parent !== undefined ? parent.thread.id : null, + workspaceStrategy: scheduledTaskWorkspaceStrategy(input.bindToCurrentThread), + }), + }) .pipe( - Effect.mapError((error) => - failure("orchestration_error", `Could not update scheduled task: ${error.message}`), - ), + Effect.catchTags({ + ScheduledTaskError: (error) => + Effect.fail( + failure( + "orchestration_error", + `Could not update scheduled task: ${error.message}`, + ), + ), + }), ); - return yield* summarizeScheduledTask(scope, { parent, limits }, task); + if (Option.isNone(updated)) { + return yield* failure( + "task_not_found", + `Scheduled task ${input.scheduledTaskId} was not found.`, + ); + } + return yield* summarizeScheduledTask(scope, { parent, limits }, updated.value.task); }), deleteScheduledTask: (scope, input) => Effect.gen(function* () { const { parent, limits } = yield* loadCaller(scope); const existing = yield* loadScheduledTask(input.scheduledTaskId, limits); yield* assertLiveCallerForOtherProject(scope, parent, existing.projectId); - yield* scheduledTasks - .delete({ id: existing.id }) + // Project ownership is enforced inside the DELETE itself, so a task + // moved to another project after authorization is a missing row — + // never a delete in a project the caller was not checked against. + const deleted = yield* scheduledTasks + .delete({ id: existing.id, projectId: existing.projectId }) .pipe( Effect.mapError((error) => failure("orchestration_error", `Could not delete scheduled task: ${error.message}`), ), ); - return { scheduledTaskId: existing.id, deleted: true }; + if (Option.isNone(deleted)) { + return yield* failure( + "task_not_found", + `Scheduled task ${input.scheduledTaskId} was not found.`, + ); + } + return { scheduledTaskId: deleted.value.id, deleted: true }; }), requestSecret: (scope, input) => Effect.gen(function* () { diff --git a/apps/server/src/mcp/OrchestratorMcpToolkit.integration.test.ts b/apps/server/src/mcp/OrchestratorMcpToolkit.integration.test.ts index aeae1a94c68b..7c6901c98d93 100644 --- a/apps/server/src/mcp/OrchestratorMcpToolkit.integration.test.ts +++ b/apps/server/src/mcp/OrchestratorMcpToolkit.integration.test.ts @@ -505,6 +505,7 @@ const layerUnusedScheduledTaskStub = Layer.succeed( list: () => Effect.succeed({ tasks: [] }), subscribeList: () => Stream.succeed({ tasks: [] }), upsert: () => Effect.die("ScheduledTaskService.upsert is unused in this test"), + update: () => Effect.die("ScheduledTaskService.update is unused in this test"), setEnabled: () => Effect.die("ScheduledTaskService.setEnabled is unused in this test"), delete: () => Effect.die("ScheduledTaskService.delete is unused in this test"), runNow: () => Effect.die("ScheduledTaskService.runNow is unused in this test"), @@ -668,12 +669,63 @@ describe("orchestrator MCP toolkit", () => { ]); return { task }; }), + update: (input) => + Ref.modify(scheduledStore, (all) => { + const existing = all.find( + (candidate) => + candidate.id === input.id && candidate.projectId === input.projectId, + ); + if (existing === undefined) return [Option.none(), all]; + const task: ScheduledTask = { + ...existing, + ...(input.title === undefined ? {} : { title: input.title }), + ...(input.prompt === undefined ? {} : { prompt: input.prompt }), + ...(input.enabled === undefined ? {} : { enabled: input.enabled }), + ...(input.schedule === undefined + ? {} + : { + schedule: + input.schedule.type === "webhook" + ? { + type: "webhook" as const, + signature: + input.schedule.signature == null + ? null + : { + header: input.schedule.signature.header, + encoding: input.schedule.signature.encoding, + prefix: input.schedule.signature.prefix, + }, + maxDeliveryAgeMinutes: + input.schedule.maxDeliveryAgeMinutes ?? null, + } + : input.schedule, + }), + ...(input.threadId === undefined ? {} : { threadId: input.threadId }), + ...(input.workspaceStrategy === undefined + ? {} + : { workspaceStrategy: input.workspaceStrategy }), + }; + return [ + Option.some({ task }), + all.map((candidate) => (candidate.id === task.id ? task : candidate)), + ]; + }), setEnabled: () => Effect.die("ScheduledTaskService.setEnabled is unused in this test"), delete: (input) => - Ref.update(scheduledStore, (all) => - all.filter((candidate) => candidate.id !== input.id), - ).pipe(Effect.as({ id: input.id })), + Ref.modify(scheduledStore, (all) => { + const existing = all.find( + (candidate) => + candidate.id === input.id && + (input.projectId === undefined || candidate.projectId === input.projectId), + ); + if (existing === undefined) return [Option.none(), all]; + return [ + Option.some({ id: existing.id }), + all.filter((candidate) => candidate.id !== existing.id), + ]; + }), runNow: () => Effect.die("ScheduledTaskService.runNow is unused in this test"), rotateWebhookToken: () => Effect.die("unused in this test"), listWebhookDeliveries: () => Effect.die("unused in this test"), @@ -1532,7 +1584,13 @@ describe("orchestrator MCP toolkit", () => { tasks: [{ scheduledTaskId, boundThreadId: parentThreadId }], }); - // update_scheduled_task pauses without deleting. + // update_scheduled_task is a partial update: the title survives an + // enabled-only edit, and pausing does not delete the task. + const scheduledTitleCall = yield* invoke("update_scheduled_task", { + scheduledTaskId, + title: "Renamed wake-up", + }); + expect(scheduledTitleCall.isError).toBe(false); const scheduledUpdateCall = yield* invoke("update_scheduled_task", { scheduledTaskId, enabled: false, @@ -1540,6 +1598,7 @@ describe("orchestrator MCP toolkit", () => { expect(scheduledUpdateCall.isError).toBe(false); expect(scheduledUpdateCall.structuredContent).toMatchObject({ scheduledTaskId, + title: "Renamed wake-up", enabled: false, }); @@ -1552,6 +1611,18 @@ describe("orchestrator MCP toolkit", () => { }); expect(yield* Ref.get(scheduledStore)).toHaveLength(0); + // An edit landing after the delete loses to a typed not-found + // rather than resurrecting the row. + const staleUpdateCall = yield* invoke("update_scheduled_task", { + scheduledTaskId, + title: "Should not resurrect", + }); + expect(declaredFailure(staleUpdateCall)).toMatchObject({ + _tag: "OrchestratorMcpFailure", + code: "task_not_found", + }); + expect(yield* Ref.get(scheduledStore)).toHaveLength(0); + // OpenCode 1.15 has emitted this exact nested-object-as-JSON-string // shape. Decode it at the MCP boundary rather than failing a task // the model otherwise specified correctly. diff --git a/apps/server/src/mcp/toolkits/core.test.ts b/apps/server/src/mcp/toolkits/core.test.ts index a9e2997282ed..6175c450adf6 100644 --- a/apps/server/src/mcp/toolkits/core.test.ts +++ b/apps/server/src/mcp/toolkits/core.test.ts @@ -21,6 +21,7 @@ import { import * as Effect from "effect/Effect"; import * as TestClock from "effect/testing/TestClock"; import * as Layer from "effect/Layer"; +import * as Option from "effect/Option"; import * as Schema from "effect/Schema"; import { McpAttachmentInput } from "./attachment/input.ts"; import { McpSchema, McpServer, Tool } from "effect/ai"; @@ -633,10 +634,12 @@ it.effect("a caller cannot rewrite a scheduled task that runs above its own mode scheduledTask("task-auto", "auto"), ], }), - upsert: (input) => - Effect.succeed({ - task: { ...(scheduledTask(input.id ?? "task-auto", "auto") as object), ...input }, - } as never), + update: ({ id, projectId: _projectId, ...patch }) => + Effect.succeed( + Option.some({ + task: { ...(scheduledTask(id, "auto") as object), ...patch }, + }) as never, + ), }), ), Layer.provide(Layer.mock(ProjectService.ProjectService)({})), diff --git a/apps/server/src/mcp/toolkits/thread/handlers.ts b/apps/server/src/mcp/toolkits/thread/handlers.ts index f54065c0824e..48a52669758c 100644 --- a/apps/server/src/mcp/toolkits/thread/handlers.ts +++ b/apps/server/src/mcp/toolkits/thread/handlers.ts @@ -80,13 +80,17 @@ export const layer = McpToolAccess.toLayer(ThreadToolkit, { Effect.gen(function* () { const scheduler = yield* ScheduledTasks.ScheduledTaskService; const { tasks } = yield* scheduler.list().pipe(Effect.mapError(unavailable)); - if (!tasks.some((task) => task.id === input.taskId)) + const listed = tasks.find((task) => task.id === input.taskId); + if (listed === undefined) return yield* new OrchestratorMcpFailure({ code: "invalid_request", message: "The scheduled task was not found.", }); + // The listed project rides into the dispatch transaction: a task moved + // to another project after this check is a missing row, never a run + // fired in a project the caller was not checked against. const { task } = yield* scheduler - .runNow({ id: input.taskId }) + .runNow({ id: input.taskId, projectId: listed.projectId }) .pipe(Effect.mapError(unavailable)); return { taskId: task.id, diff --git a/apps/server/src/observability/RpcInstrumentation.ts b/apps/server/src/observability/RpcInstrumentation.ts index 167ce8d11fff..a32b6c7a02f1 100644 --- a/apps/server/src/observability/RpcInstrumentation.ts +++ b/apps/server/src/observability/RpcInstrumentation.ts @@ -87,6 +87,7 @@ const RPC_AGGREGATES = { [WS_METHODS.scheduledTasksList]: "scheduledTasks", [WS_METHODS.scheduledTasksSubscribe]: "scheduledTasks", [WS_METHODS.scheduledTasksUpsert]: "scheduledTasks", + [WS_METHODS.scheduledTasksUpdate]: "scheduledTasks", [WS_METHODS.scheduledTasksSetEnabled]: "scheduledTasks", [WS_METHODS.scheduledTasksDelete]: "scheduledTasks", [WS_METHODS.scheduledTasksRunNow]: "scheduledTasks", diff --git a/apps/server/src/scheduledTasks/ScheduledTaskService.test.ts b/apps/server/src/scheduledTasks/ScheduledTaskService.test.ts index 1c95fecdea79..febbe3951675 100644 --- a/apps/server/src/scheduledTasks/ScheduledTaskService.test.ts +++ b/apps/server/src/scheduledTasks/ScheduledTaskService.test.ts @@ -3,12 +3,24 @@ import * as NodeUtil from "node:util"; import * as NodeCrypto from "@effect/platform-node/NodeCrypto"; import { assert, it } from "@effect/vitest"; -import { ScheduledTaskError } from "@t3tools/contracts"; +import { + CommandId, + ProjectId, + ProviderInstanceId, + ScheduledTaskError, + ScheduledTaskId, + ThreadId, + type OrchestrationV2ThreadProjection, +} from "@t3tools/contracts"; import * as DateTime from "effect/DateTime"; import * as Effect from "effect/Effect"; +import * as Fiber from "effect/Fiber"; import * as Layer from "effect/Layer"; import * as Logger from "effect/Logger"; +import * as Option from "effect/Option"; +import * as Result from "effect/Result"; import * as SqlClient from "effect/sql/SqlClient"; +import * as SqlError from "effect/sql/SqlError"; import * as Deferred from "effect/Deferred"; import * as Ref from "effect/Ref"; @@ -351,3 +363,1204 @@ it.effect( ); }).pipe(Effect.provide(SqlitePersistence.layerMemory)), ); +const updateProjectId = ProjectId.make("project:atomic-update"); +const otherProjectId = ProjectId.make("project:atomic-update-other"); +const updateTaskId = ScheduledTaskId.make("scheduled-task:atomic-update"); + +const updateTestDeps = Layer.mergeAll( + SqlitePersistence.layerMemory, + NodeCrypto.layer, + Scheduler.layer, + Layer.mock(ThreadLaunchService.ThreadLaunchService)({}), + Layer.mock(ThreadManagementService.ThreadManagementService)({}), + Layer.mock(SecretRequests.SecretRequests)({}), +); + +const updateTestLayer = ScheduledTaskService.layer.pipe(Layer.provide(updateTestDeps)); + +// Only `boundThreadId` (v2) exists as a live binding target, inside +// `updateProjectId`. `v1ThreadId` exists only in the legacy projection — +// shell reconciliation runs before writes are served, so a thread without a +// v2 row can never dispatch and must be rejected. `deletedThreadId` has a +// deleted v2 row shadowing a live v1 row — v2 is the sole authority. +const boundThreadId = ThreadId.make("thread:in-project"); +const v1ThreadId = ThreadId.make("thread:v1-only"); +const deletedThreadId = ThreadId.make("thread:deleted"); +const archivedThreadId = ThreadId.make("thread:archived"); + +// Same service plus direct SQL access to its in-memory database, for tests +// that must plant row state the public API cannot express. +const updateTestLayerWithSql = ScheduledTaskService.layer.pipe(Layer.provideMerge(updateTestDeps)); + +const seedProjectThreads = Effect.gen(function* () { + const sql = yield* SqlClient.SqlClient; + const now = "2026-09-14T00:00:00.000Z"; + yield* sql` + INSERT INTO orchestration_v2_projection_threads + (thread_id, project_id, title, default_provider, runtime_mode, interaction_mode, + created_at, updated_at, payload_json) + VALUES + (${boundThreadId}, ${updateProjectId}, 'bound', 'codex', 'full-access', 'default', + ${now}, ${now}, '{}'), + (${deletedThreadId}, ${updateProjectId}, 'deleted', 'codex', 'full-access', 'default', + ${now}, ${now}, '{}'), + (${archivedThreadId}, ${updateProjectId}, 'archived', 'codex', 'full-access', 'default', + ${now}, ${now}, '{}') + `; + yield* sql` + UPDATE orchestration_v2_projection_threads + SET deleted_at = ${now} + WHERE thread_id = ${deletedThreadId} + `; + yield* sql` + UPDATE orchestration_v2_projection_threads + SET archived_at = ${now} + WHERE thread_id = ${archivedThreadId} + `; + // v1-only rows must NOT satisfy the check — dispatch reads only the v2 + // projection, so one of them shadows the deleted v2 row to prove the v1 + // table is never consulted. + yield* sql` + INSERT INTO projection_threads + (thread_id, project_id, title, created_at, updated_at) + VALUES + (${v1ThreadId}, ${updateProjectId}, 'v1', ${now}, ${now}), + (${deletedThreadId}, ${updateProjectId}, 'v1 shadow', ${now}, ${now}) + `; +}); + +const seedTask = Effect.gen(function* () { + const tasks = yield* ScheduledTaskService.ScheduledTaskService; + const { task } = yield* tasks.upsert({ + id: updateTaskId, + title: "title original", + prompt: "prompt original", + enabled: true, + schedule: { type: "interval", everyMs: 60_000 }, + projectId: updateProjectId, + threadId: null, + workspaceStrategy: { type: "root" }, + modelSelection: { instanceId: ProviderInstanceId.make("codex"), model: "gpt-5.1-codex" }, + runtimeMode: "full-access", + interactionMode: "default", + createdBy: "agent", + creationSource: "mcp", + }); + return task; +}); + +const findSeeded = Effect.gen(function* () { + const tasks = yield* ScheduledTaskService.ScheduledTaskService; + const { tasks: all } = yield* tasks.list(); + return all.find((candidate) => candidate.id === updateTaskId); +}); + +it.effect("update keeps disjoint concurrent edits and untouched fields", () => + Effect.gen(function* () { + const tasks = yield* ScheduledTaskService.ScheduledTaskService; + const seeded = yield* seedTask; + const [first, second] = yield* Effect.all( + [ + tasks.update({ id: updateTaskId, projectId: updateProjectId, title: "title A" }), + tasks.update({ id: updateTaskId, projectId: updateProjectId, prompt: "prompt B" }), + ], + { concurrency: "unbounded" }, + ); + assert.isTrue(Option.isSome(first)); + assert.isTrue(Option.isSome(second)); + const after = yield* findSeeded; + assert.isDefined(after); + // Both disjoint edits survive — neither overwrote the other's column. + assert.equal(after!.title, "title A"); + assert.equal(after!.prompt, "prompt B"); + // Unset fields keep their seeded values. + assert.equal(after!.enabled, true); + assert.deepEqual(after!.schedule, { type: "interval", everyMs: 60_000 }); + assert.equal(after!.projectId, updateProjectId); + assert.equal(after!.createdAt, seeded.createdAt); + }).pipe(Effect.provide(updateTestLayer)), +); + +it.effect("update gives a webhook trigger its URL token and signing secret", () => + Effect.gen(function* () { + const tasks = yield* ScheduledTaskService.ScheduledTaskService; + const sql = yield* SqlClient.SqlClient; + yield* seedTask; + const credentials = sql<{ webhook_token: string | null; webhook_secret: string | null }>` + SELECT webhook_token, webhook_secret FROM scheduled_tasks WHERE task_id = ${updateTaskId} + `.pipe(Effect.map((rows) => rows[0])); + const signature = { + header: "X-Hub-Signature-256", + encoding: "hex", + prefix: "sha256=", + } as const; + + // Switching to a webhook without a stored secret must not drop the check. + const missingSecret = yield* tasks + .update({ + id: updateTaskId, + projectId: updateProjectId, + schedule: { type: "webhook", signature }, + }) + .pipe(Effect.result); + if (Result.isSuccess(missingSecret)) assert.fail("expected a missing-secret failure"); + + const switched = Option.getOrThrow( + yield* tasks.update({ + id: updateTaskId, + projectId: updateProjectId, + schedule: { type: "webhook", signature: { ...signature, secret: "s3cret" } }, + }), + ).task; + assert.deepEqual(switched.schedule, { + type: "webhook", + signature: { header: "x-hub-signature-256", encoding: "hex", prefix: "sha256=" }, + maxDeliveryAgeMinutes: null, + }); + assert.isTrue(switched.webhook?.hasSecret); + const first = yield* credentials; + assert.isString(first?.webhook_token); + assert.equal(first?.webhook_secret, "s3cret"); + + // A later trigger edit without a secret keeps the stored secret and URL. + yield* tasks.update({ + id: updateTaskId, + projectId: updateProjectId, + schedule: { type: "webhook", signature, maxDeliveryAgeMinutes: 30 }, + }); + assert.deepEqual(yield* credentials, first); + + // Leaving the webhook trigger clears both. + yield* tasks.update({ + id: updateTaskId, + projectId: updateProjectId, + schedule: { type: "interval", everyMs: 60_000 }, + }); + assert.deepEqual(yield* credentials, { webhook_token: null, webhook_secret: null }); + }).pipe(Effect.provide(updateTestLayerWithSql)), +); + +it.effect("update rejects a patch whose authorized modes were raised since", () => + Effect.gen(function* () { + const tasks = yield* ScheduledTaskService.ScheduledTaskService; + yield* seedTask; + yield* tasks.update({ id: updateTaskId, projectId: updateProjectId, runtimeMode: "auto" }); + // A caller authorized against "auto" loses the race to a raise. + yield* tasks.update({ + id: updateTaskId, + projectId: updateProjectId, + runtimeMode: "full-access", + }); + const stale = yield* tasks + .update({ + id: updateTaskId, + projectId: updateProjectId, + prompt: "Run something else", + expectedRuntimeMode: "auto", + expectedInteractionMode: "default", + }) + .pipe(Effect.result); + if (Result.isSuccess(stale)) assert.fail("expected a typed conflict"); + assert.equal(stale.failure._tag, "ScheduledTaskError"); + const after = yield* findSeeded; + assert.equal(after?.prompt, "prompt original"); + assert.equal(after?.runtimeMode, "full-access"); + // Matching pins still apply. + const current = yield* tasks.update({ + id: updateTaskId, + projectId: updateProjectId, + prompt: "Run something else", + expectedRuntimeMode: "full-access", + expectedInteractionMode: "default", + }); + assert.equal(Option.getOrThrow(current).task.prompt, "Run something else"); + }).pipe(Effect.provide(updateTestLayer)), +); + +it.effect("update patches runtimeMode without disturbing other fields", () => + Effect.gen(function* () { + const tasks = yield* ScheduledTaskService.ScheduledTaskService; + const seeded = yield* seedTask; + const result = yield* tasks.update({ + id: updateTaskId, + projectId: updateProjectId, + runtimeMode: "auto", + }); + assert.isTrue(Option.isSome(result)); + const after = yield* findSeeded; + assert.isDefined(after); + assert.equal(after!.runtimeMode, "auto"); + assert.equal(after!.title, seeded.title); + assert.equal(after!.enabled, true); + assert.equal(after!.nextRunAt, seeded.nextRunAt); + }).pipe(Effect.provide(updateTestLayer)), +); + +it.effect("concurrent schedule and enabled patches merge inside the transaction", () => + Effect.gen(function* () { + const tasks = yield* ScheduledTaskService.ScheduledTaskService; + yield* seedTask; + // Each update reads the scoped row inside its own transaction, so the + // second writer sees the first writer's committed columns: whichever + // commits last must still observe enabled=false and produce a null due + // time — never a next_run_at computed from the pre-pause snapshot. + yield* Effect.all( + [ + tasks.update({ + id: updateTaskId, + projectId: updateProjectId, + schedule: { type: "interval", everyMs: 3_600_000 }, + }), + tasks.update({ id: updateTaskId, projectId: updateProjectId, enabled: false }), + ], + { concurrency: "unbounded" }, + ); + const after = yield* findSeeded; + assert.isDefined(after); + assert.equal(after!.enabled, false); + assert.deepEqual(after!.schedule, { type: "interval", everyMs: 3_600_000 }); + assert.isNull(after!.nextRunAt); + }).pipe(Effect.provide(updateTestLayer)), +); + +it.effect("concurrent workspaceStrategyPatch edits merge into the live strategy", () => + Effect.gen(function* () { + const tasks = yield* ScheduledTaskService.ScheduledTaskService; + yield* seedTask; + // Move the seeded task to a worktree strategy carrying a field no client + // control edits (`branch`), then race two disjoint control patches: two + // editors saving baseRef and startFromOrigin must both keep their change + // instead of the second whole-object write reverting the first. + yield* tasks.update({ + id: updateTaskId, + projectId: updateProjectId, + workspaceStrategy: { + type: "worktree", + baseRef: "main", + startFromOrigin: true, + branch: "release/1.2", + }, + }); + yield* Effect.all( + [ + tasks.update({ + id: updateTaskId, + projectId: updateProjectId, + workspaceStrategyPatch: { baseRef: "develop" }, + }), + tasks.update({ + id: updateTaskId, + projectId: updateProjectId, + workspaceStrategyPatch: { startFromOrigin: false }, + }), + ], + { concurrency: "unbounded" }, + ); + const after = yield* findSeeded; + assert.isDefined(after); + assert.deepEqual(after!.workspaceStrategy, { + type: "worktree", + baseRef: "develop", + startFromOrigin: false, + branch: "release/1.2", + }); + }).pipe(Effect.provide(updateTestLayer)), +); + +it.effect("workspaceStrategyPatch drops keys foreign to a concurrently switched kind", () => + Effect.gen(function* () { + const tasks = yield* ScheduledTaskService.ScheduledTaskService; + yield* seedTask; + // The seeded strategy is `root`; a stale worktree-control patch has no + // keys that belong to it, so the merge encodes the stale edit away + // instead of resurrecting a worktree strategy over the concurrent + // kind change. + const result = yield* tasks.update({ + id: updateTaskId, + projectId: updateProjectId, + workspaceStrategyPatch: { baseRef: "develop" }, + }); + assert.isTrue(Option.isSome(result)); + const after = yield* findSeeded; + assert.isDefined(after); + assert.deepEqual(after!.workspaceStrategy, { type: "root" }); + }).pipe(Effect.provide(updateTestLayer)), +); + +it.effect("update loses to a racing delete and never recreates the task", () => + Effect.gen(function* () { + const tasks = yield* ScheduledTaskService.ScheduledTaskService; + yield* seedTask; + yield* Effect.all( + [ + tasks.update({ id: updateTaskId, projectId: updateProjectId, title: "racing edit" }), + tasks.delete({ id: updateTaskId }), + ], + { concurrency: "unbounded" }, + ); + // Whichever statement committed first, the row must stay deleted — the + // update is a targeted UPDATE that can never insert. + assert.isUndefined(yield* findSeeded); + + // Deterministic stale update after the delete: typed `none`, still absent. + yield* seedTask; + yield* tasks.delete({ id: updateTaskId }); + const stale = yield* tasks.update({ + id: updateTaskId, + projectId: updateProjectId, + title: "stale edit", + }); + assert.isTrue(Option.isNone(stale)); + assert.isUndefined(yield* findSeeded); + }).pipe(Effect.provide(updateTestLayer)), +); + +it.effect("update enforces project scope and reports missing tasks as none", () => + Effect.gen(function* () { + const tasks = yield* ScheduledTaskService.ScheduledTaskService; + yield* seedTask; + const wrongProject = yield* tasks.update({ + id: updateTaskId, + projectId: otherProjectId, + title: "cross-project edit", + }); + assert.isTrue(Option.isNone(wrongProject)); + const after = yield* findSeeded; + assert.equal(after?.title, "title original"); + + const missing = yield* tasks.update({ + id: ScheduledTaskId.make("scheduled-task:missing"), + projectId: updateProjectId, + title: "no row", + }); + assert.isTrue(Option.isNone(missing)); + }).pipe(Effect.provide(updateTestLayer)), +); + +// Gated launch barrier: lets a test park the thread dispatch between +// markRunning and the completion write, so an edit lands in a deterministic +// window rather than a racy one. +let launchBarrier: { + readonly started: Deferred.Deferred; + readonly release: Deferred.Deferred; +} | null = null; + +const gatedLaunchTestLayer = ScheduledTaskService.layer.pipe( + Layer.provide( + Layer.mergeAll( + SqlitePersistence.layerMemory, + NodeCrypto.layer, + Scheduler.layer, + Layer.mock(ThreadLaunchService.ThreadLaunchService)({ + launch: () => + Effect.gen(function* () { + const barrier = launchBarrier; + if (barrier !== null) { + yield* Deferred.succeed(barrier.started, undefined); + yield* Deferred.await(barrier.release); + } + return { + threadId: ThreadId.make("thread:scheduled-run"), + projection: {} as unknown as OrchestrationV2ThreadProjection, + resumed: false, + }; + }), + }), + Layer.mock(ThreadManagementService.ThreadManagementService)({}), + Layer.mock(SecretRequests.SecretRequests)({}), + ), + ), +); + +it.effect("run completion computes the next due time from an edit committed mid-run", () => + Effect.gen(function* () { + const tasks = yield* ScheduledTaskService.ScheduledTaskService; + yield* seedTask; + const started = yield* Deferred.make(); + const release = yield* Deferred.make(); + launchBarrier = { started, release }; + try { + const runFiber = yield* tasks.runNow({ id: updateTaskId }).pipe(Effect.forkChild); + // The dispatch is parked: the update below lands after markRunning and + // before the completion transaction, deterministically. + yield* Deferred.await(started); + const edited = yield* tasks.update({ + id: updateTaskId, + projectId: updateProjectId, + schedule: { type: "interval", everyMs: 3_600_000 }, + }); + assert.isTrue(Option.isSome(edited)); + yield* Deferred.succeed(release, undefined); + yield* Fiber.join(runFiber); + } finally { + launchBarrier = null; + } + const after = yield* findSeeded; + assert.isDefined(after); + assert.deepEqual(after!.schedule, { type: "interval", everyMs: 3_600_000 }); + assert.equal(after!.lastRunStatus, "succeeded"); + // The completion transaction re-read the row, so the new one-hour + // interval — not the pre-run 60s schedule — sets the next due time. + assert.isNotNull(after!.nextRunAt); + assert.isNotNull(after!.lastRunAt); + const nextMillis = DateTime.toEpochMillis(DateTime.makeUnsafe(after!.nextRunAt!)); + const startedMillis = DateTime.toEpochMillis(DateTime.makeUnsafe(after!.lastRunAt!)); + assert.isAtLeast(nextMillis - startedMillis, 3_500_000); + }).pipe(Effect.provide(gatedLaunchTestLayer)), +); + +it.effect("interrupted run records the failure against the edit committed mid-run", () => + Effect.gen(function* () { + const tasks = yield* ScheduledTaskService.ScheduledTaskService; + yield* seedTask; + const started = yield* Deferred.make(); + const release = yield* Deferred.make(); + launchBarrier = { started, release }; + try { + const runFiber = yield* tasks.runNow({ id: updateTaskId }).pipe(Effect.forkChild); + yield* Deferred.await(started); + const edited = yield* tasks.update({ + id: updateTaskId, + projectId: updateProjectId, + schedule: { type: "interval", everyMs: 3_600_000 }, + }); + assert.isTrue(Option.isSome(edited)); + // Interrupt while the dispatch is parked: runTask captures the + // interruption as a failed run, so the failure write must also derive + // next_run_at from the edited schedule rather than the stale snapshot. + yield* Fiber.interrupt(runFiber); + } finally { + launchBarrier = null; + } + const after = yield* findSeeded; + assert.isDefined(after); + assert.deepEqual(after!.schedule, { type: "interval", everyMs: 3_600_000 }); + assert.equal(after!.lastRunStatus, "failed"); + assert.isNotNull(after!.nextRunAt); + assert.isNotNull(after!.lastRunAt); + const nextMillis = DateTime.toEpochMillis(DateTime.makeUnsafe(after!.nextRunAt!)); + const startedMillis = DateTime.toEpochMillis(DateTime.makeUnsafe(after!.lastRunAt!)); + assert.isAtLeast(nextMillis - startedMillis, 3_500_000); + }).pipe(Effect.provide(gatedLaunchTestLayer)), +); + +it.effect("update retains the pending due time unless the schedule changes", () => + Effect.gen(function* () { + const tasks = yield* ScheduledTaskService.ScheduledTaskService; + const seeded = yield* seedTask; + assert.isNotNull(seeded.nextRunAt); + + // Move the clock forward inside the pending window: an update that + // recomputed next_run_at unconditionally would now emit a different + // timestamp, so the exact-equality checks below discriminate between + // "retained" and "recomputed". (+30s stays inside the 60s interval, so + // the task is still not due and the poller cannot fire it.) + yield* TestClock.adjust("30 seconds"); + + // Non-schedule edits keep the pending due time exactly. + const renamed = yield* tasks.update({ + id: updateTaskId, + projectId: updateProjectId, + title: "renamed", + prompt: "new prompt", + threadId: null, + workspaceStrategy: { type: "root" }, + }); + assert.isTrue(Option.isSome(renamed)); + assert.equal(Option.getOrThrow(renamed).task.nextRunAt, seeded.nextRunAt); + + // Explicitly resubmitting the current enabled flag or an equal schedule + // also retains the pending due time. + const sameEnabled = yield* tasks.update({ + id: updateTaskId, + projectId: updateProjectId, + enabled: true, + }); + assert.equal(Option.getOrThrow(sameEnabled).task.nextRunAt, seeded.nextRunAt); + const sameSchedule = yield* tasks.update({ + id: updateTaskId, + projectId: updateProjectId, + schedule: { type: "interval", everyMs: 60_000 }, + }); + assert.equal(Option.getOrThrow(sameSchedule).task.nextRunAt, seeded.nextRunAt); + + // A semantically equivalent fixed-time schedule (an explicit all-weekdays + // mask means the same as an omitted one) retains it too. The due time is + // planted as a sentinel no recompute from the current clock could + // produce — exact equality discriminates here, unlike a timestamp + // reachable from `now`, which recomputes to the same value. + const sql = yield* SqlClient.SqlClient; + const fixedTimeId = ScheduledTaskId.make("scheduled-task:fixed-time"); + yield* tasks.upsert({ + id: fixedTimeId, + title: "fixed", + prompt: "fixed prompt", + enabled: true, + schedule: { type: "fixed_time", timeOfDay: "09:30", weekdays: [0, 1, 2, 3, 4, 5, 6] }, + projectId: updateProjectId, + threadId: null, + workspaceStrategy: { type: "root" }, + modelSelection: { instanceId: ProviderInstanceId.make("codex"), model: "gpt-5.1-codex" }, + runtimeMode: "full-access", + interactionMode: "default", + createdBy: "agent", + creationSource: "mcp", + }); + const overdue = "2001-01-02T09:30:00.000Z"; + yield* sql`UPDATE scheduled_tasks SET next_run_at = ${overdue} WHERE task_id = ${fixedTimeId}`; + const sameFixed = yield* tasks.update({ + id: fixedTimeId, + projectId: updateProjectId, + schedule: { type: "fixed_time", timeOfDay: "09:30" }, + }); + assert.equal(Option.getOrThrow(sameFixed).task.nextRunAt, overdue); + + // A schedule change restarts the run clock. + const rescheduled = yield* tasks.update({ + id: updateTaskId, + projectId: updateProjectId, + schedule: { type: "interval", everyMs: 3_600_000 }, + }); + const rescheduledTask = Option.getOrThrow(rescheduled).task; + assert.isNotNull(rescheduledTask.nextRunAt); + assert.notEqual(rescheduledTask.nextRunAt, seeded.nextRunAt); + + // Disabling clears the due time; editing another field while disabled + // does not resurrect one. + const paused = yield* tasks.update({ + id: updateTaskId, + projectId: updateProjectId, + enabled: false, + }); + assert.isNull(Option.getOrThrow(paused).task.nextRunAt); + const editedWhilePaused = yield* tasks.update({ + id: updateTaskId, + projectId: updateProjectId, + title: "still paused", + }); + assert.isNull(Option.getOrThrow(editedWhilePaused).task.nextRunAt); + }).pipe(Effect.provide(updateTestLayerWithSql)), +); + +it.effect("update applies model selection and project moves patch-style", () => + Effect.gen(function* () { + const tasks = yield* ScheduledTaskService.ScheduledTaskService; + const seeded = yield* seedTask; + const moved = yield* tasks.update({ + id: updateTaskId, + projectId: updateProjectId, + modelSelection: { instanceId: ProviderInstanceId.make("claude"), model: "opus-5" }, + nextProjectId: otherProjectId, + title: "moved", + }); + const movedTask = Option.getOrThrow(moved).task; + assert.equal(movedTask.title, "moved"); + assert.deepEqual(movedTask.modelSelection, { + instanceId: ProviderInstanceId.make("claude"), + model: "opus-5", + }); + assert.equal(movedTask.projectId, otherProjectId); + // Untouched fields — including the pending due time — survive the move. + assert.equal(movedTask.prompt, seeded.prompt); + assert.equal(movedTask.nextRunAt, seeded.nextRunAt); + + // After the move, the old project scope can no longer see or edit it. + const staleScope = yield* tasks.update({ + id: updateTaskId, + projectId: updateProjectId, + title: "should not land", + }); + assert.isTrue(Option.isNone(staleScope)); + const newScope = yield* tasks.update({ + id: updateTaskId, + projectId: otherProjectId, + prompt: "new scope works", + }); + assert.isTrue(Option.isSome(newScope)); + }).pipe(Effect.provide(updateTestLayer)), +); + +it.effect("scoped delete loses to a racing project move instead of deleting cross-project", () => + Effect.gen(function* () { + const tasks = yield* ScheduledTaskService.ScheduledTaskService; + yield* seedTask; + // The task moves to project B before a delete authorized for project A + // commits: the scoped DELETE finds no row in A and must not touch B's row. + const moved = yield* tasks.update({ + id: updateTaskId, + projectId: updateProjectId, + nextProjectId: otherProjectId, + }); + assert.isTrue(Option.isSome(moved)); + const staleDelete = yield* tasks.delete({ + id: updateTaskId, + projectId: updateProjectId, + }); + assert.isTrue(Option.isNone(staleDelete)); + const after = yield* findSeeded; + assert.isDefined(after); + assert.equal(after?.projectId, otherProjectId); + + // Scoped to the live project the same delete succeeds; an unscoped delete + // of the now-missing row reports none instead of a false success. + const liveDelete = yield* tasks.delete({ + id: updateTaskId, + projectId: otherProjectId, + }); + assert.isTrue(Option.isSome(liveDelete)); + assert.isUndefined(yield* findSeeded); + const goneDelete = yield* tasks.delete({ id: updateTaskId }); + assert.isTrue(Option.isNone(goneDelete)); + }).pipe(Effect.provide(updateTestLayer)), +); + +it.effect("scoped runNow loses to a racing project move and never dispatches", () => + Effect.gen(function* () { + const tasks = yield* ScheduledTaskService.ScheduledTaskService; + yield* seedTask; + dispatchLaunchCount = 0; + // A manual run authorized for project A commits after the task moved to + // project B: the dispatch transaction's re-read sees B, so the scoped run + // is a typed not-found and no launch fires under A's authority. + const moved = yield* tasks.update({ + id: updateTaskId, + projectId: updateProjectId, + nextProjectId: otherProjectId, + }); + assert.isTrue(Option.isSome(moved)); + const staleRun = yield* tasks + .runNow({ id: updateTaskId, projectId: updateProjectId }) + .pipe(Effect.result); + assert.isTrue(Result.isFailure(staleRun)); + assert.equal(dispatchLaunchCount, 0); + + // Scoped to the live project the same manual run dispatches normally. + const liveRun = yield* tasks.runNow({ + id: updateTaskId, + projectId: otherProjectId, + }); + assert.equal(liveRun.task.lastRunStatus, "succeeded"); + assert.equal(dispatchLaunchCount, 1); + }).pipe(Effect.provide(gatedDispatchLayer)), +); + +it.effect( + "legacy explicit-id upsert can move an unbound task without restarting its schedule", + () => + Effect.gen(function* () { + const tasks = yield* ScheduledTaskService.ScheduledTaskService; + const seeded = yield* seedTask; + const moved = yield* tasks.upsert({ + ...seeded, + requireExisting: true, + projectId: otherProjectId, + title: "Moved by an older client", + }); + assert.equal(moved.task.projectId, otherProjectId); + assert.equal(moved.task.title, "Moved by an older client"); + assert.equal(moved.task.nextRunAt, seeded.nextRunAt); + const stored = yield* findSeeded; + assert.equal(stored?.projectId, otherProjectId); + assert.equal(stored?.nextRunAt, seeded.nextRunAt); + }).pipe(Effect.provide(updateTestLayer)), +); + +it.effect("update keeps an unchanged binding to a since-archived thread editable", () => + Effect.gen(function* () { + const tasks = yield* ScheduledTaskService.ScheduledTaskService; + const sql = yield* SqlClient.SqlClient; + yield* seedProjectThreads; + yield* seedTask; + yield* tasks.update({ + id: updateTaskId, + projectId: updateProjectId, + threadId: boundThreadId, + enabled: false, + }); + yield* sql` + UPDATE orchestration_v2_projection_threads + SET archived_at = '2026-09-15T00:00:00.000Z' + WHERE thread_id = ${boundThreadId} + `; + // A stale draft repeats the stored binding and project with a rename. + const renamed = yield* tasks.update({ + id: updateTaskId, + projectId: updateProjectId, + threadId: boundThreadId, + nextProjectId: updateProjectId, + title: "renamed after archive", + }); + assert.equal(Option.getOrThrow(renamed).task.title, "renamed after archive"); + // A changed binding is still validated. + const rebound = yield* tasks + .update({ id: updateTaskId, projectId: updateProjectId, threadId: deletedThreadId }) + .pipe(Effect.result); + if (Result.isSuccess(rebound)) assert.fail("expected a typed conflict"); + assert.equal(rebound.failure._tag, "ScheduledTaskError"); + }).pipe(Effect.provide(updateTestLayerWithSql)), +); + +it.effect("legacy upsert keeps an unchanged binding to a since-archived thread editable", () => + Effect.gen(function* () { + const tasks = yield* ScheduledTaskService.ScheduledTaskService; + const sql = yield* SqlClient.SqlClient; + yield* seedProjectThreads; + const seeded = yield* seedTask; + const bound = yield* tasks.update({ + id: updateTaskId, + projectId: updateProjectId, + threadId: boundThreadId, + enabled: false, + }); + assert.isTrue(Option.isSome(bound)); + yield* sql` + UPDATE orchestration_v2_projection_threads + SET archived_at = '2026-09-15T00:00:00.000Z' + WHERE thread_id = ${boundThreadId} + `; + // An older client resends the stored binding with a title edit. + const legacySave = { + ...seeded, + enabled: false, + threadId: boundThreadId, + requireExisting: true, + }; + const renamed = yield* tasks.upsert({ ...legacySave, title: "renamed after archive" }); + assert.equal(renamed.task.title, "renamed after archive"); + assert.equal(renamed.task.threadId, boundThreadId); + // A changed binding is still validated. + const rebound = yield* tasks + .upsert({ ...legacySave, threadId: deletedThreadId }) + .pipe(Effect.result); + if (Result.isSuccess(rebound)) assert.fail("expected a typed conflict"); + assert.equal(rebound.failure._tag, "ScheduledTaskError"); + }).pipe(Effect.provide(updateTestLayerWithSql)), +); + +it.effect( + "replayed commandId upsert loses to a racing project move instead of dragging it back", + () => + Effect.gen(function* () { + const tasks = yield* ScheduledTaskService.ScheduledTaskService; + const replayId = ScheduledTaskId.make("scheduled-task:cmd:replay-move"); + const replayInput = { + title: "replay title", + prompt: "replay prompt", + enabled: true, + schedule: { type: "interval", everyMs: 60_000 } as const, + threadId: null, + workspaceStrategy: { type: "root" } as const, + modelSelection: { instanceId: ProviderInstanceId.make("codex"), model: "gpt-5.1-codex" }, + runtimeMode: "full-access" as const, + interactionMode: "default" as const, + createdBy: "agent" as const, + creationSource: "mcp" as const, + commandId: CommandId.make("cmd:replay-move"), + }; + const { task: created } = yield* tasks.upsert({ + ...replayInput, + projectId: updateProjectId, + }); + assert.equal(created.id, replayId); + + // The task moves to project B before a stale MCP replay of the same + // schedule_task call lands: the resolved id still matches, but the scoped + // upsert must answer not-found rather than rewrite the row back into A. + const moved = yield* tasks.update({ + id: replayId, + projectId: updateProjectId, + nextProjectId: otherProjectId, + }); + assert.isTrue(Option.isSome(moved)); + const replay = yield* tasks + .upsert({ ...replayInput, projectId: updateProjectId }) + .pipe(Effect.result); + assert.isTrue(Result.isFailure(replay)); + const { tasks: all } = yield* tasks.list(); + const after = all.find((candidate) => candidate.id === replayId); + assert.isDefined(after); + assert.equal(after?.projectId, otherProjectId); + assert.equal(after?.title, created.title); + + // A replay authorized for the live project is still the idempotent write. + const liveReplay = yield* tasks.upsert({ ...replayInput, projectId: otherProjectId }); + assert.equal(liveReplay.task.projectId, otherProjectId); + }).pipe(Effect.provide(updateTestLayer)), +); + +// ---------- deterministic scheduler probes ---------- +// Post-statement hooks on the service's SqlClient pin the contested windows: +// the due-read hook runs after the SELECT's connection permit is released, so +// a competing edit commits in exactly the read/dispatch gap; the mark-read +// hook proves the dispatch transaction revalidated the row; a follow-up +// due-read gives a sleep-free endpoint after which the iteration is done. +type SqlProbe = ( + statement: string, + rows: ReadonlyArray, +) => Effect.Effect | void; + +let sqlProbe: SqlProbe | null = null; + +const gatedSqlClient = Layer.unwrap( + Effect.gen(function* () { + const inner = yield* SqlClient.SqlClient; + const call = inner as ( + strings: TemplateStringsArray, + ...args: ReadonlyArray + ) => Effect.Effect, SqlError.SqlError>; + const gated = Object.assign((strings: unknown, ...args: ReadonlyArray): unknown => { + if (typeof strings === "string") { + return (inner as (name: string) => unknown)(strings); + } + const statement = call(strings as TemplateStringsArray, ...args); + const effect = statement.pipe( + Effect.andThen((rows) => { + const probe = sqlProbe; + if (probe === null) return Effect.succeed(rows); + const text = (strings as TemplateStringsArray).join(" "); + const hook = probe(text, rows); + return hook === undefined ? Effect.succeed(rows) : Effect.as(hook, rows); + }), + ); + // Piping drops the Fragment interface, which `sql.and` clauses rely on + // to embed — restore the data props so the gated value stays both an + // executable Effect and an embeddable fragment. + return Object.assign(effect, { + "~effect/sql/Fragment": "~effect/sql/Fragment", + segments: (statement as unknown as { segments: ReadonlyArray }).segments, + }); + }, inner); + return Layer.succeed(SqlClient.SqlClient, gated as unknown as SqlClient.SqlClient); + }), +); + +let dispatchLaunchCount = 0; +let dispatchLaunched: Deferred.Deferred | null = null; + +const countingLaunchLayer = Layer.mock(ThreadLaunchService.ThreadLaunchService)({ + launch: () => + Effect.gen(function* () { + dispatchLaunchCount += 1; + const signal = dispatchLaunched; + if (signal !== null) yield* Deferred.succeed(signal, undefined); + return { + threadId: ThreadId.make("thread:scheduled-run"), + projection: {} as unknown as OrchestrationV2ThreadProjection, + resumed: false, + }; + }), +}); + +// A lock timeout reported only through a raw `errcode`, without the client's +// `errno` normalization: the native-code fallback must still retry it. +const nodeBusyError = () => + new SqlError.SqlError({ + reason: new SqlError.UnknownError({ + cause: Object.assign(new Error("database is locked"), { errcode: 5 }), + }), + }); + +// The same lock timeout as Bun reports it — `code`/`errno`, which the +// classifier maps to a retryable LockTimeoutError. Both shapes must retry. +const bunBusyError = () => + new SqlError.SqlError({ + reason: new SqlError.LockTimeoutError({ + cause: Object.assign(new Error("database is locked"), { + code: "SQLITE_BUSY", + errno: 5, + }), + }), + }); + +const gatedDispatchLayer = ScheduledTaskService.layer.pipe( + Layer.provideMerge( + Layer.mergeAll( + gatedSqlClient.pipe(Layer.provide(SqlitePersistence.layerMemory)), + NodeCrypto.layer, + Scheduler.layer, + countingLaunchLayer, + Layer.mock(ThreadManagementService.ThreadManagementService)({}), + Layer.mock(SecretRequests.SecretRequests)({}), + ), + ), +); + +const isDueRead = (statement: string) => + statement.includes("FROM scheduled_tasks") && statement.includes("next_run_at <="); +const isMarkRead = (statement: string) => + statement.includes("FROM scheduled_tasks") && + statement.includes("WHERE task_id") && + !statement.includes("AND project_id") && + !statement.includes("next_run_at <="); +const isTerminalWrite = (statement: string) => + statement.includes("UPDATE scheduled_tasks") && statement.includes("run_count = run_count + 1"); + +// Arms the due-read gate: the first non-empty due read parks the scheduler +// between "found due" and "dispatch"; every later due read ticks nextPoll so +// the test can wait for the iteration to fully drain without sleeping. +const armDueReadGate = ( + dueArrived: Deferred.Deferred, + dueProceed: Deferred.Deferred, + nextPoll: Deferred.Deferred, + rechecked: Deferred.Deferred, +) => { + let dueSeen = false; + sqlProbe = (statement, rows) => { + if (isDueRead(statement)) { + if (!dueSeen && rows.length > 0) { + dueSeen = true; + return Deferred.succeed(dueArrived, undefined).pipe( + Effect.andThen(Deferred.await(dueProceed)), + ); + } + if (dueSeen) return Deferred.succeed(nextPoll, undefined); + return; + } + if (dueSeen && isMarkRead(statement)) { + return Deferred.succeed(rechecked, undefined); + } + }; +}; + +// Each edit commits while the poll that found the task due is parked before +// the dispatch transaction revalidates the row; the run must honour it. +it.effect.each([ + { + name: "pause", + edit: (tasks: ScheduledTaskService.ScheduledTaskService["Service"]) => + tasks.update({ id: updateTaskId, projectId: updateProjectId, enabled: false }), + }, + { + name: "postpone", + edit: (tasks: ScheduledTaskService.ScheduledTaskService["Service"]) => + tasks.update({ + id: updateTaskId, + projectId: updateProjectId, + schedule: { type: "interval", everyMs: 3_600_000 }, + }), + }, + { + name: "delete", + edit: (tasks: ScheduledTaskService.ScheduledTaskService["Service"]) => + tasks.delete({ id: updateTaskId }), + }, +])("scheduled dispatch honours a $name committed after the due read", ({ edit }) => + Effect.gen(function* () { + const tasks = yield* ScheduledTaskService.ScheduledTaskService; + yield* seedTask; // interval 60s: due at T+60s + const dueArrived = yield* Deferred.make(); + const dueProceed = yield* Deferred.make(); + const rechecked = yield* Deferred.make(); + const nextPoll = yield* Deferred.make(); + dispatchLaunchCount = 0; + dispatchLaunched = null; + armDueReadGate(dueArrived, dueProceed, nextPoll, rechecked); + try { + const adjust = yield* TestClock.adjust("65 seconds").pipe(Effect.forkChild); + yield* Deferred.await(dueArrived); + yield* edit(tasks); + const edited = yield* findSeeded; + yield* Deferred.succeed(dueProceed, undefined); + yield* Fiber.join(adjust); + yield* Deferred.await(rechecked); // the mark transaction re-read the row + yield* TestClock.adjust("5 seconds"); + yield* Deferred.await(nextPoll); // the contested iteration fully drained + assert.equal(dispatchLaunchCount, 0); + const after = yield* findSeeded; + // The skipped dispatch must not stamp over the edit's state, including + // a postponed due time. + assert.equal(after?.nextRunAt, edited?.nextRunAt); + assert.equal(after?.lastRunStatus, edited?.lastRunStatus); + if (edited !== undefined) assert.equal(after?.lastRunStatus, "never"); + } finally { + sqlProbe = null; + } + }).pipe(Effect.provide(gatedDispatchLayer)), +); + +it.effect("scheduled dispatch still fires when no edit lands in the gap", () => + Effect.gen(function* () { + yield* seedTask; + const dueArrived = yield* Deferred.make(); + const dueProceed = yield* Deferred.make(); + const rechecked = yield* Deferred.make(); + const nextPoll = yield* Deferred.make(); + const settled = yield* Deferred.make(); + const launched = yield* Deferred.make(); + dispatchLaunchCount = 0; + dispatchLaunched = launched; + let dueSeen = false; + sqlProbe = (statement, rows) => { + if (isDueRead(statement)) { + if (!dueSeen && rows.length > 0) { + dueSeen = true; + return Deferred.succeed(dueArrived, undefined).pipe( + Effect.andThen(Deferred.await(dueProceed)), + ); + } + if (dueSeen) return Deferred.succeed(nextPoll, undefined); + return; + } + if (dueSeen && isMarkRead(statement)) { + return Deferred.succeed(rechecked, undefined); + } + if (isTerminalWrite(statement)) { + return Deferred.succeed(settled, undefined); + } + }; + try { + const adjust = yield* TestClock.adjust("65 seconds").pipe(Effect.forkChild); + yield* Deferred.await(dueArrived); + yield* Deferred.succeed(dueProceed, undefined); + yield* Fiber.join(adjust); + yield* Deferred.await(rechecked); + yield* Deferred.await(launched); + yield* Deferred.await(settled); + assert.equal(dispatchLaunchCount, 1); + const after = yield* findSeeded; + assert.equal(after!.lastRunStatus, "succeeded"); + assert.equal(after!.runCount, 1); + } finally { + sqlProbe = null; + dispatchLaunched = null; + } + }).pipe(Effect.provide(gatedDispatchLayer)), +); + +it.effect("a contended completion write retries instead of stranding the task", () => + Effect.gen(function* () { + const tasks = yield* ScheduledTaskService.ScheduledTaskService; + yield* seedTask; + dispatchLaunchCount = 0; + let injected = false; + sqlProbe = (statement) => { + // One-shot lock timeout on the completion write: without the + // contention retry the transaction's failure propagates and the row is + // left last_run_status='running', which due scans then exclude forever. + if (!injected && isTerminalWrite(statement) && statement.includes("last_run_at =")) { + injected = true; + return Effect.fail(nodeBusyError()); + } + }; + try { + const ran = yield* tasks.runNow({ id: updateTaskId }); + assert.isTrue(injected); // the probe fired — the retry path ran + assert.equal(ran.task.lastRunStatus, "succeeded"); + assert.equal(ran.task.runCount, 1); + } finally { + sqlProbe = null; + } + }).pipe(Effect.provide(gatedDispatchLayer)), +); + +it.effect("a contended recovery write retries instead of stranding the task", () => + Effect.gen(function* () { + const tasks = yield* ScheduledTaskService.ScheduledTaskService; + yield* seedTask; + dispatchLaunchCount = 0; + let completionFails = 0; + let releaseFails = 0; + sqlProbe = (statement) => { + if (!isTerminalWrite(statement)) return; + // Fail every completion write past the retry budget so the onError + // recovery path runs, then fail its terminal write once. Without the + // contention retry that leaves the row 'running' — excluded from due + // scans forever. ('failed' is a literal only in releaseStuckRun's SQL.) + if (statement.includes("last_run_at =")) { + completionFails += 1; + return Effect.fail(nodeBusyError()); + } + if (statement.includes("'failed'")) { + releaseFails += 1; + // Bun-style contention shape — exercises the isRetryable branch. + if (releaseFails === 1) return Effect.fail(bunBusyError()); + } + }; + try { + const ran = yield* tasks.runNow({ id: updateTaskId }).pipe(Effect.result); + assert.isTrue(Result.isFailure(ran)); // the exhausted completion tx surfaces + const after = yield* findSeeded; + assert.isAbove(completionFails, 1); // the completion retry really ran + assert.isAbove(releaseFails, 0); // the recovery write was contested + assert.equal(after!.lastRunStatus, "failed"); + assert.equal(after!.runCount, 1); + assert.isNotNull(after!.nextRunAt); // re-aimed, not dropped + } finally { + sqlProbe = null; + } + }).pipe(Effect.provide(gatedDispatchLayer)), +); + +it.effect("update keeps a project move and a thread binding consistent", () => + Effect.gen(function* () { + const tasks = yield* ScheduledTaskService.ScheduledTaskService; + yield* seedTask; + yield* seedProjectThreads; + // Binding to a v2 thread that lives in the task's project works. + const bound = yield* tasks.update({ + id: updateTaskId, + projectId: updateProjectId, + threadId: boundThreadId, + }); + assert.isTrue(Option.isSome(bound)); + // A v1-only thread cannot be bound: the v2 projection is the only + // authority dispatch reads, and it has no row for this thread. + const boundV1 = yield* tasks + .update({ id: updateTaskId, projectId: updateProjectId, threadId: v1ThreadId }) + .pipe(Effect.result); + if (Result.isSuccess(boundV1)) assert.fail("expected a typed conflict"); + assert.equal(boundV1.failure._tag, "ScheduledTaskError"); + // A deleted v2 row still wins over a live v1 row for the same thread. + const deleted = yield* tasks + .update({ id: updateTaskId, projectId: updateProjectId, threadId: deletedThreadId }) + .pipe(Effect.result); + if (Result.isSuccess(deleted)) assert.fail("expected a typed conflict"); + assert.equal(deleted.failure._tag, "ScheduledTaskError"); + // An archived thread is rejected the same way: sendToThread refuses + // archived targets, so a bound task would save as runnable and then fail + // every dispatch. + const archived = yield* tasks + .update({ id: updateTaskId, projectId: updateProjectId, threadId: archivedThreadId }) + .pipe(Effect.result); + if (Result.isSuccess(archived)) assert.fail("expected a typed conflict"); + assert.equal(archived.failure._tag, "ScheduledTaskError"); + // The contested merge: a move patch landing on top of the committed + // binding cannot keep a thread from the old project — the merged pair + // would fail every dispatch, so it is a typed conflict instead. + const moved = yield* tasks + .update({ id: updateTaskId, projectId: updateProjectId, nextProjectId: otherProjectId }) + .pipe(Effect.result); + if (Result.isSuccess(moved)) assert.fail("expected a typed conflict"); + assert.equal(moved.failure._tag, "ScheduledTaskError"); + const kept = yield* findSeeded; + assert.equal(kept?.projectId, updateProjectId); + assert.equal(kept?.threadId, boundThreadId); + // The explicit escape hatch: unbind and move in the same patch. + const unbound = yield* tasks.update({ + id: updateTaskId, + projectId: updateProjectId, + nextProjectId: otherProjectId, + threadId: null, + }); + const movedTask = Option.getOrThrow(unbound).task; + assert.isNull(movedTask.threadId); + assert.equal(movedTask.projectId, otherProjectId); + // The reverse interleaving: after the move, binding a thread from the old + // project conflicts too. + const relink = yield* tasks + .update({ id: updateTaskId, projectId: otherProjectId, threadId: boundThreadId }) + .pipe(Effect.result); + if (Result.isSuccess(relink)) assert.fail("expected a typed conflict"); + assert.equal(relink.failure._tag, "ScheduledTaskError"); + // And binding a thread that does not exist in the project at all. + const missing = yield* tasks + .update({ + id: updateTaskId, + projectId: otherProjectId, + threadId: ThreadId.make("thread:nowhere"), + }) + .pipe(Effect.result); + if (Result.isSuccess(missing)) assert.fail("expected a typed conflict"); + assert.equal(missing.failure._tag, "ScheduledTaskError"); + }).pipe(Effect.provide(updateTestLayerWithSql)), +); diff --git a/apps/server/src/scheduledTasks/ScheduledTaskService.ts b/apps/server/src/scheduledTasks/ScheduledTaskService.ts index 51403c124a98..1328be6b6c21 100644 --- a/apps/server/src/scheduledTasks/ScheduledTaskService.ts +++ b/apps/server/src/scheduledTasks/ScheduledTaskService.ts @@ -2,6 +2,7 @@ import { CommandId, MessageId, PROVIDER_SEND_TURN_MAX_INPUT_CHARS, + ProjectId, ScheduledTask, ScheduledTaskError, ScheduledTaskId, @@ -20,6 +21,7 @@ import { type ScheduledTaskRunNowInput, type ScheduledTaskRunNowResult, type ScheduledTaskSetEnabledInput, + type ScheduledTaskUpdateInput, type ScheduledTaskUpsertInput, } from "@t3tools/contracts"; import * as Cause from "effect/Cause"; @@ -32,6 +34,7 @@ import * as Duration from "effect/Duration"; import * as Metric from "effect/Metric"; import * as Layer from "effect/Layer"; import * as Option from "effect/Option"; +import * as Predicate from "effect/Predicate"; import * as PubSub from "effect/PubSub"; import * as Ref from "effect/Ref"; import * as Result from "effect/Result"; @@ -39,6 +42,7 @@ import * as Schema from "effect/Schema"; import * as Semaphore from "effect/Semaphore"; import * as Stream from "effect/Stream"; import * as SqlClient from "effect/sql/SqlClient"; +import * as SqlError from "effect/sql/SqlError"; import * as ThreadLaunchService from "../orchestration-v2/ThreadLaunchService.ts"; import * as Metrics from "../observability/Metrics.ts"; @@ -173,6 +177,16 @@ const decodeHeadersJson = Schema.decodeUnknownOption(HeadersJson); const decodeMissingFieldsJson = Schema.decodeUnknownOption(MissingFieldsJson); const encodeHeadersJson = Schema.encodeSync(HeadersJson); const encodeMissingFieldsJson = Schema.encodeSync(MissingFieldsJson); +const encodeScheduleJson = Schema.encodeEffect( + Schema.fromJsonString(ScheduledTask.fields.schedule), +); +const encodeWorkspaceStrategyJson = Schema.encodeEffect( + Schema.fromJsonString(ScheduledTask.fields.workspaceStrategy), +); +const encodeModelSelectionJson = Schema.encodeEffect( + Schema.fromJsonString(ScheduledTask.fields.modelSelection), +); +const isScheduledTaskError = Schema.is(ScheduledTaskError); interface ScheduledTaskRow { readonly task_id: string; @@ -225,13 +239,27 @@ export class ScheduledTaskService extends Context.Service< readonly upsert: ( input: ScheduledTaskUpsertInput, ) => Effect.Effect; + /** + * Atomic partial update: writes only the provided fields, and only while + * the task still exists in `input.projectId`. Returns `Option.none()` when + * no such task exists — the update can never insert a row, so an edit + * racing a delete loses instead of resurrecting the task. + * + * A `threadId`/`nextProjectId` patch is a typed conflict when the merged + * pair would bind the task to a thread outside its project (the dispatch + * check in `getProjectThread` would reject it). + * Unbind explicitly with `threadId: null` to move a bound task. + */ + readonly update: ( + input: ScheduledTaskUpdateInput, + ) => Effect.Effect, ScheduledTaskError>; /** Partial update flipping only the enabled flag; never touches other fields. */ readonly setEnabled: ( input: ScheduledTaskSetEnabledInput, ) => Effect.Effect; readonly delete: ( input: ScheduledTaskDeleteInput, - ) => Effect.Effect; + ) => Effect.Effect, ScheduledTaskError>; readonly runNow: ( input: ScheduledTaskRunNowInput, ) => Effect.Effect; @@ -291,6 +319,34 @@ function errorMessage(error: unknown): string { return String(error); } +// Write transactions take the lock up front and wait up to busy_timeout, so +// contention only surfaces when another process holds the lock longer than +// that. Bun reports `code`/`errno`, and the node:sqlite client copies +// `errcode` into `errno`, so both normally arrive as a retryable +// LockTimeoutError. Accept that classification, and as a fallback a raw +// native code in the SQLITE_BUSY (5) / SQLITE_LOCKED (6) families on any of +// the field names a driver might use without that normalization. The ScheduledTaskError unwrap covers statements +// that wrap SqlError inside the transaction before the retry policy sees it. +const isContendedWriteError = (cause: unknown): boolean => { + if (isScheduledTaskError(cause)) return isContendedWriteError(cause.cause); + if (!SqlError.isSqlError(cause)) return false; + if (cause.reason.isRetryable) return true; + const native = cause.reason.cause; + for (const key of ["errcode", "errno", "code"] as const) { + if (Predicate.hasProperty(native, key) && typeof native[key] === "number") { + const base = native[key] & 0xff; + if (base === 5 || base === 6) return true; + } + } + return false; +}; + +// Retry a write that timed out waiting for another process's lock, so a +// lock held a little past busy_timeout does not drop an edit or strand a run +// as 'running'. +const retryContended = (effect: Effect.Effect) => + effect.pipe(Effect.retry({ times: 2, while: isContendedWriteError })); + function webhookPath(taskId: string, token: string): string { return `${WEBHOOK_ROUTE_PREFIX}/${encodeURIComponent(taskId)}/${token}`; } @@ -476,6 +532,35 @@ export const layer = Layer.effect( WHERE task_id = ${id} `; + const getScopedRows = (id: ScheduledTaskId, projectId: ScheduledTask["projectId"]) => + sql` + 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, + webhook_token, + webhook_secret + FROM scheduled_tasks + WHERE task_id = ${id} AND project_id = ${projectId} + `; + /** 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( @@ -500,6 +585,26 @@ export const layer = Layer.effect( ), ); + // Stored webhook schedules keep only the read-model fields: the signing + // secret lives in its own column and the header matches case-insensitively. + const normalizeSchedule = ( + schedule: ScheduledTaskUpsertInput["schedule"], + ): ScheduledTask["schedule"] => + schedule.type === "webhook" + ? { + type: "webhook", + signature: + schedule.signature == null + ? null + : { + header: schedule.signature.header.toLowerCase(), + encoding: schedule.signature.encoding, + prefix: schedule.signature.prefix, + }, + maxDeliveryAgeMinutes: schedule.maxDeliveryAgeMinutes ?? null, + } + : schedule; + const newWebhookToken = crypto.randomBytes(32).pipe( Effect.map((bytes) => Buffer.from(bytes).toString("base64url")), Effect.mapError((cause) => taskError("Could not generate webhook token.", { cause })), @@ -513,6 +618,43 @@ export const layer = Layer.effect( return task; }); + // A task only ever dispatches through getProjectThread, which reads the + // v2 projection and requires the bound thread to live in the task's + // project and not be deleted — so any stored (project, thread) pair that + // check would reject is a task that can never fire. This is the + // side-effect-free mirror of that rule: a plain read of the same v2 + // projection table, safe to run inside the update transaction because it + // never takes the importer's thread lock or hydrates a transcript a + // rollback would orphan. (No v1 fallback: startup shell reconciliation + // has already imported every v1 thread before writes are served, and + // ensureTranscript never creates shells, so a thread without a v2 row is + // one dispatch can never reach.) + const requireThreadInProject = Effect.fn("ScheduledTaskService.requireThreadInProject")( + function* (taskId: ScheduledTaskId, projectId: ProjectId, threadId: ThreadId) { + const counts = yield* sql<{ matched: number }>` + SELECT COUNT(*) AS matched + FROM orchestration_v2_projection_threads + WHERE thread_id = ${threadId} + AND project_id = ${projectId} + AND deleted_at IS NULL + AND archived_at IS NULL + `.pipe( + Effect.mapError((cause) => + taskError("Could not validate the schedule task's thread binding.", { + taskId, + cause, + }), + ), + ); + if ((counts[0]?.matched ?? 0) === 0) { + return yield* taskError( + "The task's thread binding is not a live thread in the task's project; unbind it or rebind to a thread in the project.", + { taskId }, + ); + } + }, + ); + // 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. @@ -615,11 +757,28 @@ export const layer = Layer.effect( ), ); - const deleteRow = (id: ScheduledTaskId) => + // Scoped like update: when the caller supplies a project, the DELETE's + // WHERE carries it, so a task moved to another project mid-request is a + // missing row, never a cross-project delete. RETURNING surfaces that. The + // task's webhook delivery log goes with it in the same transaction. + const deleteRow = (id: ScheduledTaskId, projectId?: ProjectId) => sql .withTransaction( - sql`DELETE FROM scheduled_task_webhook_deliveries WHERE task_id = ${id}`.pipe( - Effect.andThen(sql`DELETE FROM scheduled_tasks WHERE task_id = ${id}`), + (projectId === undefined + ? sql<{ + task_id: string; + }>`DELETE FROM scheduled_tasks WHERE task_id = ${id} RETURNING task_id` + : sql<{ + task_id: string; + }>`DELETE FROM scheduled_tasks WHERE task_id = ${id} AND project_id = ${projectId} RETURNING task_id` + ).pipe( + Effect.flatMap((rows) => + rows.length > 0 + ? sql`DELETE FROM scheduled_task_webhook_deliveries WHERE task_id = ${id}`.pipe( + Effect.as(rows), + ) + : Effect.succeed(rows), + ), ), ) .pipe( @@ -685,18 +844,41 @@ export const layer = Layer.effect( // 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' - `; + // Re-read and write under one lock so an edit committed between them + // keeps its own next_run_at rather than being overwritten by the + // schedule snapshot taken before the edit. Contention retry keeps a + // lock held by another process from stranding the task as 'running'. + yield* retryContended( + sql + .withTransaction( + Effect.gen(function* () { + 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' + `; + }), + ) + .pipe( + Effect.catchTags({ + SqlError: (cause) => + Effect.fail( + taskError("Could not release stuck schedule task run.", { + taskId: task.id, + cause, + }), + ), + }), + ), + ); yield* notifyChanged; }).pipe( Effect.catch((cause) => @@ -711,6 +893,7 @@ export const layer = Layer.effect( task: ScheduledTask, trigger: "scheduled" | "manual" | "webhook", webhook?: { readonly deliveryId: string; readonly prompt: string }, + projectId?: ProjectId, ) { const reserved = yield* Ref.modify(activeRuns, (active) => { if (active.has(task.id)) return [false, active] as const; @@ -729,11 +912,67 @@ export const layer = Layer.effect( 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) { + // The in-memory snapshot may be stale: re-read and mark running under + // one transaction, so a pause, postpone, or delete committed between + // the poll's read and this dispatch still wins instead of firing a + // run the edit just invalidated. + const marked = yield* retryContended( + sql + .withTransaction( + Effect.gen(function* () { + const active = yield* findTask(task.id); + // A scoped manual run re-reads inside the transaction that + // marks the task running, so a task moved to another project + // since authorization is a missing row — never a dispatch + // under the old project's authority. + if (active === null || (projectId !== undefined && active.projectId !== projectId)) + return null; + // A next_run_at corrupted between the poll read and this + // re-read must not defect the poll; an unparseable value is + // treated as not due. + const parsedNextRunAt = + active.nextRunAt === null ? Option.none() : DateTime.make(active.nextRunAt); + if ( + trigger === "scheduled" && + (!active.enabled || + Option.isNone(parsedNextRunAt) || + DateTime.toEpochMillis(parsedNextRunAt.value) > + DateTime.toEpochMillis(startedAt)) + ) { + return { task: active, running: false } as const; + } + // A queued delivery must not run a task that was paused, + // deleted and recreated under the same id, or switched to + // another trigger while it waited for its turn. + if (webhook !== undefined) { + const reason = + active.createdAt !== task.createdAt + ? "The task was replaced before this delivery ran." + : active.schedule.type !== "webhook" + ? "The task's trigger changed before this delivery ran." + : !active.enabled + ? "The task was paused before this delivery ran." + : null; + if (reason !== null) return yield* new WebhookDeliverySkipped({ reason }); + } + yield* markRunning(active.id, startedAtIso); + return { task: active, running: true } as const; + }), + ) + .pipe( + Effect.catchTags({ + ScheduledTaskError: (error) => Effect.fail(error), + SqlError: (cause) => + Effect.fail( + taskError("Could not mark schedule task as running.", { + taskId: task.id, + cause, + }), + ), + }), + ), + ); + if (marked === null) { if (webhook !== undefined) { return yield* new WebhookDeliverySkipped({ reason: "The task was deleted before this delivery ran.", @@ -746,34 +985,8 @@ export const layer = Layer.effect( } return task; } - // A next_run_at corrupted between the poll read and this re-read must - // not defect the poll; an unparseable value is treated as not due. - const parsedNextRunAt = - active.nextRunAt === null ? Option.none() : DateTime.make(active.nextRunAt); - if ( - trigger === "scheduled" && - (!active.enabled || - Option.isNone(parsedNextRunAt) || - DateTime.toEpochMillis(parsedNextRunAt.value) > DateTime.toEpochMillis(startedAt)) - ) { - return active; - } - // A queued delivery must not run a task that was paused, deleted and - // recreated under the same id, or switched to another trigger while - // it waited for its turn. - if (webhook !== undefined) { - const reason = - active.createdAt !== task.createdAt - ? "The task was replaced before this delivery ran." - : active.schedule.type !== "webhook" - ? "The task's trigger changed before this delivery ran." - : !active.enabled - ? "The task was paused before this delivery ran." - : null; - if (reason !== null) return yield* new WebhookDeliverySkipped({ reason }); - } - - yield* markRunning(active.id, startedAtIso); + if (!marked.running) return marked.task; + const active = marked.task; yield* notifyChanged; // A webhook run is keyed by its delivery so the same delivery can @@ -835,8 +1048,50 @@ export const layer = Layer.effect( const lastRunStatus = runSucceeded ? ("succeeded" as const) : ("failed" as const); const lastRunError = runSucceeded ? null : errorMessage(result.cause); // Re-read the task so the next run is computed from the schedule as it - // is *now* (the user may have edited or deleted it while we ran). - const current = yield* findTask(task.id); + // is *now* (the user may have edited or deleted it while we ran). The + // re-read and the write share one transaction so a concurrent edit + // cannot land between them and have its next_run_at overwritten by a + // stale schedule. Contention retry keeps a lock held by another + // process from abandoning the completion write with the task left + // 'running'. + const current = yield* retryContended( + sql + .withTransaction( + Effect.gen(function* () { + const current = yield* findTask(task.id); + if (current !== null) { + // startedAtIso in the guard ensures this writes only to the row + // this run marked as running — a task deleted mid-run and + // recreated with the same id (idempotent commandId replay) must + // not be stamped. + yield* markCompleted({ + id: task.id, + completedAtIso: iso(completedAt), + nextRunAtIso: nextRunAt(current, completedAt), + status: lastRunStatus, + error: lastRunError, + startedAtIso, + }); + } + return current; + }), + ) + .pipe( + Effect.catchTags({ + ScheduledTaskError: (error) => Effect.fail(error), + SqlError: (cause) => + Effect.fail( + taskError("Could not record schedule task run.", { + taskId: task.id, + cause, + }), + ), + }), + ), + ); + if (current !== null) { + yield* notifyChanged; + } const scheduleSource = current ?? task; const completed: ScheduledTask = { ...scheduleSource, @@ -847,20 +1102,6 @@ export const layer = Layer.effect( lastRunError, runCount: scheduleSource.runCount + 1, }; - if (current !== null) { - // startedAtIso in the guard ensures this writes only to the row this - // run marked as running — a task deleted mid-run and recreated with - // the same id (idempotent commandId replay) must not be stamped. - yield* markCompleted({ - id: task.id, - completedAtIso: completed.updatedAt, - nextRunAtIso: completed.nextRunAt, - status: lastRunStatus, - error: lastRunError, - startedAtIso, - }); - yield* notifyChanged; - } return completed; }).pipe( Effect.onError((cause) => releaseStuckRun(task, errorMessage(cause))), @@ -886,11 +1127,14 @@ export const layer = Layer.effect( missedRunAt: task.nextRunAt, rescheduledTo: next, }); + // Guard on the due time we read: an edit that rescheduled the task + // since the poll changes next_run_at, so this compare-and-swap loses + // instead of overwriting the edit with a stale computation. yield* sql` UPDATE scheduled_tasks SET next_run_at = ${next}, updated_at = ${iso(now)} - WHERE task_id = ${task.id} + WHERE task_id = ${task.id} AND next_run_at = ${task.nextRunAt} `.pipe( Effect.mapError((cause) => taskError("Could not reschedule missed schedule task run.", { taskId: task.id, cause }), @@ -941,7 +1185,7 @@ export const layer = Layer.effect( Effect.gen(function* () { const decoded = yield* Effect.result(decodeRow(row)); if (Result.isSuccess(decoded)) { - yield* sql` + yield* retryContended(sql` UPDATE scheduled_tasks SET last_run_status = 'failed', last_run_error = 'Run was interrupted by a server restart.', @@ -949,7 +1193,7 @@ export const layer = Layer.effect( updated_at = ${iso(now)}, run_count = run_count + 1 WHERE task_id IS ${row.task_id} AND last_run_status = 'running' - `; + `); return; } // The schedule cannot be decoded, so the next occurrence cannot @@ -959,14 +1203,14 @@ export const layer = Layer.effect( "Recovering undecodable schedule task row without rescheduling", { taskId: row.task_id, cause: decoded.failure }, ); - yield* sql` + yield* retryContended(sql` UPDATE scheduled_tasks SET last_run_status = 'failed', last_run_error = 'Run was interrupted by a server restart.', updated_at = ${iso(now)}, run_count = run_count + 1 WHERE task_id IS ${row.task_id} AND last_run_status = 'running' - `; + `); }), { concurrency: 1, discard: true }, ); @@ -1000,161 +1244,387 @@ export const layer = Layer.effect( const upsert: ScheduledTaskService["Service"]["upsert"] = (input) => Effect.gen(function* () { - const now = yield* localNow; - const uuid = - input.commandId === undefined - ? yield* crypto.randomUUIDv4.pipe( + const signatureInput = input.schedule.type === "webhook" ? input.schedule.signature : null; + // A secretRef is a value the user entered for an agent; this save + // consumes it, so it cannot be used again. Consume it once, outside + // the write transaction, so a contended-write retry cannot spend it + // twice. + const consumedRef = + signatureInput?.secretRef === undefined + ? undefined + : yield* Effect.result( + secretRequests.consume({ + ref: signatureInput.secretRef, + projectId: input.projectId, + }), + ); + const task = yield* retryContended( + sql.withTransaction( + Effect.gen(function* () { + const now = yield* localNow; + const uuid = + input.commandId === undefined + ? yield* crypto.randomUUIDv4.pipe( + Effect.mapError((cause) => + taskError("Could not generate schedule task id.", { cause }), + ), + ) + : null; + const id = + input.id ?? + ScheduledTaskId.make( + input.commandId ? `scheduled-task:${input.commandId}` : `scheduled-task:${uuid}`, + ); + // Look up by the *resolved* id so idempotent creates (commandId replays) + // keep their run history, and so real load failures propagate instead + // of silently resetting an existing row. + const existingTask = yield* findTask(id); + // A replayed create resolves the same id even after the task + // left the caller's project — the scoped view must answer + // not-found rather than let the upsert drag the row back and + // overwrite a definition the caller can no longer see. + if ( + input.id === undefined && + existingTask !== null && + existingTask.projectId !== input.projectId + ) { + return yield* taskError("Schedule task not found.", { taskId: id }); + } + const schedule = normalizeSchedule(input.schedule); + const webhook = + input.schedule.type === "webhook" + ? yield* Effect.gen(function* () { + const existing = + existingTask === null ? null : yield* findWebhookCredentials(id); + // Saving a webhook task keeps its URL; only rotate changes it. + const token = existing?.token ?? (yield* newWebhookToken); + // A replay of a save that already used the ref keeps + // the secret that save stored. + const replay = input.commandId !== undefined && existing?.secret != null; + const fromRef = + consumedRef === undefined + ? undefined + : Result.isSuccess(consumedRef) + ? consumedRef.success + : replay + ? undefined + : yield* taskError(consumedRef.failure.message, { taskId: id }); + // A ref, when given, is the only source: a plain secret sent + // alongside it must not replace what a replayed save stored. + const provided = + signatureInput?.secretRef === undefined ? signatureInput?.secret : fromRef; + const secret = + signatureInput == null ? null : (provided ?? existing?.secret ?? null); + const secretChanged = + signatureInput == null || provided !== undefined || existing === null; + if (signatureInput != null && secret === null) { + return yield* taskError( + "A webhook signature check needs a signing secret.", + { taskId: id }, + ); + } + return { token, secret, secretChanged }; + }) + : { token: null, secret: null, secretChanged: true }; + // Keep the existing next_run_at when the schedule itself is untouched: + // editing a title or prompt must not postpone (or resurrect) a due + // run — only schedule/enabled changes restart the clock. + const scheduleUnchanged = + existingTask !== null && + existingTask.enabled === input.enabled && + isSameSchedule(existingTask.schedule, schedule); + const task: ScheduledTask = { + id, + title: input.title, + prompt: input.prompt, + enabled: input.enabled, + schedule, + projectId: input.projectId, + threadId: input.threadId ?? null, + workspaceStrategy: input.workspaceStrategy, + modelSelection: input.modelSelection, + runtimeMode: input.runtimeMode, + interactionMode: input.interactionMode, + createdBy: existingTask?.createdBy ?? input.createdBy ?? "user", + creationSource: input.creationSource ?? "web", + createdAt: existingTask?.createdAt ?? iso(now), + updatedAt: iso(now), + nextRunAt: scheduleUnchanged + ? existingTask.nextRunAt + : nextRunAt({ enabled: input.enabled, schedule }, now), + lastRunAt: existingTask?.lastRunAt ?? null, + lastRunStatus: existingTask?.lastRunStatus ?? "never", + lastRunError: existingTask?.lastRunError ?? null, + runCount: existingTask?.runCount ?? 0, + }; + // Like update, validate only a binding this write changes: a + // legacy full-row save or a commandId replay resends the stored + // pair, and its thread being archived since must not block an + // unrelated title or prompt edit. + const bindingChanged = + existingTask === null || + existingTask.threadId !== task.threadId || + existingTask.projectId !== task.projectId; + if (task.threadId !== null && bindingChanged) { + yield* requireThreadInProject(task.id, task.projectId, task.threadId); + } + yield* saveTask(task, input.requireExisting === true, webhook); + return task; + }), + ), + ).pipe( + Effect.catchTags({ + ScheduledTaskError: (error) => Effect.fail(error), + SqlError: (cause) => + Effect.fail(taskError("Could not upsert the schedule task.", { cause })), + }), + ); + yield* notifyChanged; + // Re-read so the result carries the webhook endpoint the row stores. + return { task: (yield* findTask(task.id)) ?? task }; + }); + + // Partial edits run inside one transaction: the scoped row is read and + // rewritten without another fiber's statements interleaving (the client's + // connection serializes transactions), so a delete racing the edit + // surfaces as `none` (the UPDATE never inserts) and concurrent disjoint + // edits each keep their own columns. + const update: ScheduledTaskService["Service"]["update"] = (input) => + Effect.gen(function* () { + const schedule = + input.schedule === undefined ? undefined : normalizeSchedule(input.schedule); + const signatureInput = input.schedule?.type === "webhook" ? input.schedule.signature : null; + // Consumed once, before the write transaction, so a contended-write + // retry cannot spend the ref twice. + const consumedRef = + signatureInput?.secretRef === undefined + ? undefined + : yield* secretRequests + .consume({ ref: signatureInput.secretRef, projectId: input.projectId }) + .pipe(Effect.mapError((error) => taskError(error.message, { taskId: input.id }))); + const task = yield* retryContended( + sql.withTransaction( + Effect.gen(function* () { + const rows = yield* getScopedRows(input.id, input.projectId).pipe( Effect.mapError((cause) => - taskError("Could not generate schedule task id.", { cause }), + taskError("Could not load schedule task.", { taskId: input.id, cause }), ), - ) - : null; - const id = - input.id ?? - ScheduledTaskId.make( - input.commandId ? `scheduled-task:${input.commandId}` : `scheduled-task:${uuid}`, - ); - // Look up by the *resolved* id so idempotent creates (commandId replays) - // keep their run history, and so real load failures propagate instead - // of silently resetting an existing row. - const existingTask = yield* findTask(id); - // Keep the existing next_run_at when the schedule itself is untouched: - // editing a title or prompt must not postpone (or resurrect) a due - // run — only schedule/enabled changes restart the clock. - const schedule: ScheduledTask["schedule"] = - input.schedule.type === "webhook" - ? { - type: "webhook", - signature: - input.schedule.signature == null - ? null - : { - header: input.schedule.signature.header.toLowerCase(), - encoding: input.schedule.signature.encoding, - prefix: input.schedule.signature.prefix, - }, - maxDeliveryAgeMinutes: input.schedule.maxDeliveryAgeMinutes ?? null, + ); + const row = rows[0]; + if (row === undefined) return null; + const existing = yield* decodeRow(row, yield* readWebhookOrigin); + if ( + (input.expectedRuntimeMode !== undefined && + existing.runtimeMode !== input.expectedRuntimeMode) || + (input.expectedInteractionMode !== undefined && + existing.interactionMode !== input.expectedInteractionMode) + ) { + return yield* taskError( + "Scheduled task changed since it was loaded; retry the operation.", + { taskId: input.id }, + ); } - : input.schedule; - const webhook = - input.schedule.type === "webhook" - ? yield* Effect.gen(function* () { - const existing = existingTask === null ? null : yield* findWebhookCredentials(id); - // Saving a webhook task keeps its URL; only rotate changes it. - const token = existing?.token ?? (yield* newWebhookToken); - const signature = - input.schedule.type === "webhook" ? input.schedule.signature : null; - // A secretRef is a value the user entered for an agent; this - // save consumes it, so it cannot be used again. A replay of a - // save that already used it keeps the secret that save stored. - const replay = input.commandId !== undefined && existing?.secret != null; - const fromRef = - signature?.secretRef === undefined - ? undefined - : yield* secretRequests - .consume({ ref: signature.secretRef, projectId: input.projectId }) - .pipe( - Effect.catch((error) => - replay - ? Effect.succeed(undefined) - : Effect.fail(taskError(error.message, { taskId: id })), - ), - ); - // A ref, when given, is the only source: a plain secret sent - // alongside it must not replace what a replayed save stored. - const provided = signature?.secretRef === undefined ? signature?.secret : fromRef; - const secret = signature == null ? null : (provided ?? existing?.secret ?? null); - const secretChanged = - signature == null || provided !== undefined || existing === null; - if (signature != null && secret === null) { - return yield* taskError("A webhook signature check needs a signing secret.", { - taskId: id, - }); + // Patches merge with whatever concurrent edits already + // committed, so the resulting pair must still be dispatchable: + // a move keeps an existing threadId only when that thread + // belongs to the destination project, and a binding update + // keeps the current project. Passing `threadId: null` in the + // same patch is the explicit unbind that lets a bound task move. + if (input.threadId !== undefined || input.nextProjectId !== undefined) { + const mergedThreadId = + input.threadId === undefined ? existing.threadId : input.threadId; + const mergedProjectId = input.nextProjectId ?? existing.projectId; + // Like upsert, only a pair this write changes is validated, so + // a patch that repeats the stored binding of a since-archived + // thread does not lose the unrelated edits it carries. + const bindingChanged = + mergedThreadId !== existing.threadId || mergedProjectId !== existing.projectId; + if (mergedThreadId !== null && bindingChanged) { + yield* requireThreadInProject(input.id, mergedProjectId, mergedThreadId); } - return { token, secret, secretChanged }; - }) - : { token: null, secret: null, secretChanged: true }; - const scheduleUnchanged = - existingTask !== null && - existingTask.enabled === input.enabled && - isSameSchedule(existingTask.schedule, schedule); - const task: ScheduledTask = { - id, - title: input.title, - prompt: input.prompt, - enabled: input.enabled, - schedule, - projectId: input.projectId, - threadId: input.threadId ?? null, - workspaceStrategy: input.workspaceStrategy, - modelSelection: input.modelSelection, - runtimeMode: input.runtimeMode, - interactionMode: input.interactionMode, - createdBy: existingTask?.createdBy ?? input.createdBy ?? "user", - creationSource: input.creationSource ?? "web", - createdAt: existingTask?.createdAt ?? iso(now), - updatedAt: iso(now), - nextRunAt: scheduleUnchanged - ? existingTask.nextRunAt - : nextRunAt({ enabled: input.enabled, schedule }, now), - lastRunAt: existingTask?.lastRunAt ?? null, - lastRunStatus: existingTask?.lastRunStatus ?? "never", - lastRunError: existingTask?.lastRunError ?? null, - runCount: existingTask?.runCount ?? 0, - }; - yield* saveTask(task, input.requireExisting === true, webhook); + } + const patch: Record = {}; + if (input.title !== undefined) patch.title = input.title; + if (input.prompt !== undefined) patch.prompt = input.prompt; + if (input.enabled !== undefined) patch.enabled = input.enabled ? 1 : 0; + if (schedule !== undefined) { + patch.schedule_json = yield* encodeScheduleJson(schedule); + if (schedule.type === "webhook") { + // Same rules as upsert: the URL token survives every save + // (only rotate changes it), and an omitted secret keeps the + // stored one. + const provided = + signatureInput?.secretRef === undefined ? signatureInput?.secret : consumedRef; + const secret = + signatureInput == null ? null : (provided ?? row.webhook_secret ?? null); + if (signatureInput != null && secret === null) { + return yield* taskError("A webhook signature check needs a signing secret.", { + taskId: input.id, + }); + } + patch.webhook_token = row.webhook_token ?? (yield* newWebhookToken); + patch.webhook_secret = secret; + } else { + patch.webhook_token = null; + patch.webhook_secret = null; + } + } + if (input.threadId !== undefined) patch.thread_id = input.threadId; + if (input.workspaceStrategy !== undefined) { + patch.workspace_strategy_json = yield* encodeWorkspaceStrategyJson( + input.workspaceStrategy, + ); + } else if (input.workspaceStrategyPatch !== undefined) { + // A sparse patch merges into the strategy read above inside + // this transaction: two clients editing disjoint controls + // (baseRef vs startFromOrigin) both land instead of the + // second whole-object write reverting the first. Keys that + // do not belong to the live strategy kind encode away, so a + // patch racing a kind change drops the stale control edit. + const sparse = input.workspaceStrategyPatch; + patch.workspace_strategy_json = yield* encodeWorkspaceStrategyJson({ + ...existing.workspaceStrategy, + ...(sparse.branch !== undefined ? { branch: sparse.branch } : {}), + ...(sparse.worktreePath !== undefined + ? { worktreePath: sparse.worktreePath } + : {}), + ...(sparse.baseRef !== undefined ? { baseRef: sparse.baseRef } : {}), + ...(sparse.startFromOrigin !== undefined + ? { startFromOrigin: sparse.startFromOrigin } + : {}), + }); + } + if (input.modelSelection !== undefined) { + patch.model_selection_json = yield* encodeModelSelectionJson(input.modelSelection); + } + if (input.runtimeMode !== undefined) patch.runtime_mode = input.runtimeMode; + if (input.nextProjectId !== undefined) patch.project_id = input.nextProjectId; + if (Object.keys(patch).length === 0) return existing; + const now = yield* localNow; + // Mirror the upsert rule: only a real schedule/enabled change + // restarts the run clock — other edits retain the pending due + // time. + if (input.enabled !== undefined || schedule !== undefined) { + const nextEnabled = input.enabled ?? existing.enabled; + const nextSchedule = schedule ?? existing.schedule; + if ( + nextEnabled !== existing.enabled || + !isSameSchedule(existing.schedule, nextSchedule) + ) { + patch.next_run_at = nextRunAt( + { enabled: nextEnabled, schedule: nextSchedule }, + now, + ); + } + } + patch.updated_at = iso(now); + // The read above and this write share one write-locked + // transaction, so no other write can land between them. + const written = yield* sql` + UPDATE scheduled_tasks + SET ${sql.update(patch)} + WHERE task_id = ${input.id} AND project_id = ${input.projectId} + RETURNING * + `.pipe( + Effect.mapError((cause) => + taskError("Could not update schedule task.", { taskId: input.id, cause }), + ), + ); + const updatedRow = written[0]; + if (updatedRow === undefined) return null; + return yield* decodeRow(updatedRow, yield* readWebhookOrigin); + }), + ), + ).pipe( + Effect.catchTags({ + ScheduledTaskError: (error) => Effect.fail(error), + SqlError: (cause) => + Effect.fail( + taskError("Could not update schedule task.", { taskId: input.id, cause }), + ), + SchemaError: (cause) => + Effect.fail( + taskError("Could not update schedule task.", { taskId: input.id, cause }), + ), + }), + ); + if (task === null) return Option.none(); yield* notifyChanged; - return { task: (yield* findTask(id)) ?? task }; + return Option.some({ task }); }); const setEnabled: ScheduledTaskService["Service"]["setEnabled"] = (input) => Effect.gen(function* () { - const existing = yield* loadTask(input.id); - if (existing.enabled === input.enabled) return { task: existing }; - const now = yield* localNow; - const next = nextRunAt({ enabled: input.enabled, schedule: existing.schedule }, now); - // RETURNING so a task deleted between the load and this UPDATE is a - // visible not-found error, not a false success. - const updated = yield* sql<{ task_id: string }>` - UPDATE scheduled_tasks - SET enabled = ${input.enabled ? 1 : 0}, - next_run_at = ${next}, - updated_at = ${iso(now)} - WHERE task_id = ${input.id} - RETURNING task_id - `.pipe( - Effect.mapError((cause) => - taskError("Could not update schedule task.", { taskId: input.id, cause }), + // Read and write share one transaction: the recomputed next_run_at can + // never be based on a schedule a concurrent edit has already replaced. + const outcome = yield* retryContended( + sql.withTransaction( + Effect.gen(function* () { + const rows = yield* getRows(input.id).pipe( + Effect.mapError((cause) => + taskError("Could not load schedule task.", { taskId: input.id, cause }), + ), + ); + const row = rows[0]; + if (row === undefined) return null; + const existing = yield* decodeRow(row, yield* readWebhookOrigin); + if (existing.enabled === input.enabled) { + return { task: existing, changed: false } as const; + } + const now = yield* localNow; + const next = nextRunAt({ enabled: input.enabled, schedule: existing.schedule }, now); + const updated = yield* sql<{ task_id: string }>` + UPDATE scheduled_tasks + SET enabled = ${input.enabled ? 1 : 0}, + next_run_at = ${next}, + updated_at = ${iso(now)} + WHERE task_id = ${input.id} + RETURNING task_id + `.pipe( + Effect.mapError((cause) => + taskError("Could not update schedule task.", { taskId: input.id, cause }), + ), + ); + if (updated.length === 0) return null; + return { + task: { ...existing, enabled: input.enabled, nextRunAt: next, updatedAt: iso(now) }, + changed: true, + } as const; + }), ), + ).pipe( + Effect.catchTags({ + ScheduledTaskError: (error) => Effect.fail(error), + SqlError: (cause) => + Effect.fail( + taskError("Could not update schedule task.", { taskId: input.id, cause }), + ), + }), ); - if (updated.length === 0) { + if (outcome === null) { return yield* taskError("Schedule task not found.", { taskId: input.id }); } - yield* notifyChanged; - return { - task: { ...existing, enabled: input.enabled, nextRunAt: next, updatedAt: iso(now) }, - }; + if (outcome.changed) yield* notifyChanged; + return { task: outcome.task }; }); const deleteTask: ScheduledTaskService["Service"]["delete"] = (input) => - deleteRow(input.id).pipe( - Effect.andThen( - Effect.all([ - Ref.update(webhookRateWindows, (windows) => { - const next = new Map(windows); - next.delete(input.id); - return next; - }), - Ref.update(webhookPermits, (permits) => { - const next = new Map(permits); - next.delete(input.id); - return next; - }), - ]), - ), - Effect.andThen(notifyChanged), - Effect.as({ id: input.id }), - ); + Effect.gen(function* () { + const rows = yield* deleteRow(input.id, input.projectId); + if (rows.length === 0) return Option.none(); + yield* Ref.update(webhookRateWindows, (windows) => { + const next = new Map(windows); + next.delete(input.id); + return next; + }); + yield* Ref.update(webhookPermits, (permits) => { + const next = new Map(permits); + next.delete(input.id); + return next; + }); + yield* notifyChanged; + return Option.some({ id: input.id }); + }); const runNow: ScheduledTaskService["Service"]["runNow"] = (input: ScheduledTaskRunNowInput) => Effect.gen(function* () { @@ -1165,7 +1635,7 @@ export const layer = Layer.effect( taskId: input.id, }); } - const next = yield* runTask(task, "manual").pipe( + const next = yield* runTask(task, "manual", undefined, input.projectId).pipe( Effect.mapError((cause) => taskError("Could not run schedule task.", { taskId: input.id, cause }), ), @@ -1663,6 +2133,7 @@ export const layer = Layer.effect( list, subscribeList, upsert, + update, setEnabled, delete: deleteTask, runNow, diff --git a/apps/server/src/ws.ts b/apps/server/src/ws.ts index 375c2ac78aa1..370640925f44 100644 --- a/apps/server/src/ws.ts +++ b/apps/server/src/ws.ts @@ -90,6 +90,7 @@ import { PersistChatAttachmentsError, RpcClientId, EnvironmentAuthorizationError, + ScheduledTaskError, type ProjectId, type ProviderDriverKind, type ProviderInstanceId, @@ -2056,6 +2057,20 @@ const layerWsRpc = ( [WS_METHODS.scheduledTasksSubscribe]: (_input) => scheduledTasks.subscribeList().pipe(Stream.map(withVisibleWebhookUrls)), [WS_METHODS.scheduledTasksUpsert]: (input) => scheduledTasks.upsert(input), + [WS_METHODS.scheduledTasksUpdate]: (input) => + Effect.annotateCurrentSpan({ "scheduled_task.id": input.id }).pipe( + Effect.andThen(scheduledTasks.update(input)), + Effect.flatMap((updated) => + Option.isNone(updated) + ? Effect.fail( + new ScheduledTaskError({ + message: "Schedule task not found in this project.", + taskId: input.id, + }), + ) + : Effect.succeed(updated.value), + ), + ), [WS_METHODS.scheduledTasksSetEnabled]: (input) => Effect.annotateCurrentSpan({ "scheduled_task.id": input.id }).pipe( Effect.andThen(scheduledTasks.setEnabled(input)), @@ -2063,6 +2078,16 @@ const layerWsRpc = ( [WS_METHODS.scheduledTasksDelete]: (input) => Effect.annotateCurrentSpan({ "scheduled_task.id": input.id }).pipe( Effect.andThen(scheduledTasks.delete(input)), + Effect.flatMap((deleted) => + Option.isNone(deleted) + ? Effect.fail( + new ScheduledTaskError({ + message: "Schedule task not found.", + taskId: input.id, + }), + ) + : Effect.succeed(deleted.value), + ), ), [WS_METHODS.scheduledTasksRunNow]: (input) => Effect.annotateCurrentSpan({ "scheduled_task.id": input.id }).pipe( diff --git a/apps/web/src/components/settings/ScheduledTasksSettings.tsx b/apps/web/src/components/settings/ScheduledTasksSettings.tsx index 992bccd78bc0..eb2194e24036 100644 --- a/apps/web/src/components/settings/ScheduledTasksSettings.tsx +++ b/apps/web/src/components/settings/ScheduledTasksSettings.tsx @@ -13,7 +13,6 @@ import { type ReactNode, useCallback, useEffect, useMemo, useRef, useState } fro import type { EnvironmentId, ModelSelection, - OrchestrationV2ThreadLaunchWorkspaceStrategy, ProjectId, ScheduledTask, ScheduledTaskId, @@ -26,6 +25,7 @@ import type { import { DEFAULT_WEBHOOK_PROMPT } from "@t3tools/client-runtime/scheduled-task-webhook"; import { MAX_WEBHOOK_DELIVERY_AGE_MINUTES, + scheduledTaskLegacyUpsert, MIN_SCHEDULED_TASK_INTERVAL_MS, ProviderInstanceId, resolveEnvironmentMachineKind, @@ -72,6 +72,12 @@ import { type ScheduleMode, type WorkspaceMode, } from "./scheduledTasksSettings.logic"; +import { + buildScheduledTaskUpdateInput, + moveDetachesThreadBinding, + workspaceStrategyFromDraft, + workspaceStrategyReady, +} from "./scheduledTasks.logic"; import { Label } from "../ui/label"; import { Menu, MenuTrigger, MenuPopup, MenuItem, MenuSeparator } from "../ui/menu"; import { ToggleGroup, Toggle } from "../ui/toggle-group"; @@ -146,7 +152,7 @@ function Field({ children, }: { label: string; - hint?: string; + hint?: string | undefined; htmlFor?: string; children: ReactNode; }) { @@ -804,9 +810,16 @@ function ScheduledTaskEditorDialog({ const upsertTask = useAtomCommand(serverEnvironment.upsertScheduledTask, { label: "scheduled task upsert", }); + const updateTask = useAtomCommand(serverEnvironment.updateScheduledTask, { + label: "scheduled task update", + }); const [draft, setDraft] = useState(() => task ? taskToDraft(task) : { ...EMPTY_DRAFT, projectId: projects[0]?.id ?? "" }, ); + // The draft as the editor opened it. Dirty detection compares the draft + // against this snapshot — not the live task — so edits another client + // commits while the dialog is open are never reverted by a save. + const [baselineDraft] = useState(draft); const [saving, setSaving] = useState(false); const submissionPending = useRef(false); const editingTaskMissing = @@ -906,7 +919,7 @@ function ScheduledTaskEditorDialog({ reportFailure("Invalid interval", "Enter an interval of at least one minute."); return; } - if (draft.workspaceMode === "existing_worktree" && !draft.existingWorktreePath.trim()) { + if (!workspaceStrategyReady(draft)) { reportFailure("Checkout path is required", "Enter the path of the checkout to run in."); return; } @@ -918,36 +931,56 @@ function ScheduledTaskEditorDialog({ draft.baseModelSelection.model === selection.model ? draft.baseModelSelection : selection; - const workspaceStrategy: OrchestrationV2ThreadLaunchWorkspaceStrategy = - draft.workspaceMode === "root" - ? { type: "root" } - : draft.workspaceMode === "existing_worktree" - ? { type: "existing_worktree", worktreePath: draft.existingWorktreePath.trim() } - : { - type: "worktree", - baseRef: draft.baseRef.trim() || "main", - startFromOrigin: draft.startFromOrigin, - }; - const input: ScheduledTaskUpsertInput = { - ...(draft.editingId ? { id: draft.editingId as ScheduledTaskId, requireExisting: true } : {}), - title: draft.title.trim(), - prompt: draft.prompt.trim(), - enabled: draft.enabled, - schedule, - projectId: selectedProjectId as ProjectId, - threadId: draft.threadId ? (draft.threadId as ThreadId) : null, - workspaceStrategy, - modelSelection, - runtimeMode: draft.runtimeMode, - interactionMode: draft.interactionMode, - creationSource: "web", - }; + const workspaceStrategy = workspaceStrategyFromDraft(draft); // Lock before React renders, and keep successful creates locked until the form closes. submissionPending.current = true; setSaving(true); - const result = await upsertTask({ environmentId, input }); + // Existing tasks save as a dirty-field patch through the atomic update + // path: a stale editor can never overwrite fields another client changed, + // and a delete racing the save is a typed not-found, not a resurrection. + let result; + if (draft.editingId !== null) { + // editingTaskMissing already guarantees the task is in the live list. + const editingTask = tasksQuery.data.tasks.find((entry) => entry.id === draft.editingId); + if (editingTask === undefined) { + submissionPending.current = false; + setSaving(false); + return; + } + const patch = buildScheduledTaskUpdateInput( + draft, + baselineDraft, + editingTask, + modelSelection, + workspaceStrategy, + ); + result = + patch === null + ? null + : environment?.serverConfig?.environment.capabilities.scheduledTaskUpdate === true + ? await updateTask({ environmentId, input: patch }) + : await upsertTask({ + environmentId, + input: scheduledTaskLegacyUpsert(editingTask, patch), + }); + } else { + const input: ScheduledTaskUpsertInput = { + title: draft.title.trim(), + prompt: draft.prompt.trim(), + enabled: draft.enabled, + schedule, + projectId: selectedProjectId as ProjectId, + threadId: draft.threadId ? (draft.threadId as ThreadId) : null, + workspaceStrategy, + modelSelection, + runtimeMode: draft.runtimeMode, + interactionMode: draft.interactionMode, + creationSource: "web", + }; + result = await upsertTask({ environmentId, input }); + } setSaving(false); - if (result._tag === "Failure") { + if (result !== null && result._tag === "Failure") { submissionPending.current = false; if (!isAtomCommandInterrupted(result)) { reportFailure("Could not save scheduled task", squashAtomCommandFailure(result)); @@ -1043,7 +1076,19 @@ function ScheduledTaskEditorDialog({
- + entry.id === draft.editingId), + ) + ? "detaches the thread binding" + : undefined + } + htmlFor="scheduled-task-project" + >