From d6134f8a4bbc56f87d773fbf54f362b1e870b27f Mon Sep 17 00:00:00 2001 From: 3xcalibur Date: Thu, 5 Nov 2026 00:58:20 +0400 Subject: [PATCH 1/8] feat(scheduled-tasks): bound tasks can steer, steer-or-start, or restart instead of queueing Scheduled and webhook tasks bound to a thread always queued behind the running turn. Tasks now store a delivery mode (queue, auto, steer, restart) that the run passes to sendToThread. It defaults to queue, and a save that omits it keeps the stored mode. Web and mobile editors show it for thread-bound tasks. Co-Authored-By: Claude Opus 5.5 --- .../SettingsScheduledTasksRouteScreen.tsx | 30 ++++++++++++++++ .../features/settings/scheduledTaskDraft.ts | 5 +++ .../src/mcp/OrchestratorMcpService.test.ts | 2 ++ apps/server/src/persistence/Migrations.ts | 2 ++ .../Migrations/080_ScheduledTaskSendMode.ts | 7 ++++ .../reconcileV2PreviewMigration.test.ts | 2 ++ .../ScheduledTaskService.test.ts | 15 +++++--- .../scheduledTasks/ScheduledTaskService.ts | 11 +++++- .../ScheduledTasksSettings.browser.test.tsx | 1 + .../settings/ScheduledTasksSettings.tsx | 36 +++++++++++++++++++ .../scheduledTasksSettings.logic.test.ts | 1 + .../settings/scheduledTasksSettings.logic.ts | 3 ++ docs/user/project-settings.md | 8 +++++ packages/client-runtime/package.json | 4 +++ .../src/scheduledTaskSendMode.ts | 27 ++++++++++++++ packages/contracts/src/scheduledTask.ts | 9 +++++ 16 files changed, 158 insertions(+), 5 deletions(-) create mode 100644 apps/server/src/persistence/Migrations/080_ScheduledTaskSendMode.ts create mode 100644 packages/client-runtime/src/scheduledTaskSendMode.ts diff --git a/apps/mobile/src/features/settings/SettingsScheduledTasksRouteScreen.tsx b/apps/mobile/src/features/settings/SettingsScheduledTasksRouteScreen.tsx index 32f05c21020d8..727943456ce6b 100644 --- a/apps/mobile/src/features/settings/SettingsScheduledTasksRouteScreen.tsx +++ b/apps/mobile/src/features/settings/SettingsScheduledTasksRouteScreen.tsx @@ -54,6 +54,10 @@ import { ScreenScrollView as ScrollView } from "../../components/ui/ScreenScroll import { SegmentedControl } from "../../components/ui/SegmentedControl"; import { ThemedSwitch } from "../../components/ui/ThemedSwitch"; import { webhookAddress } from "@t3tools/client-runtime/webhook-address"; +import { + SCHEDULED_TASK_SEND_MODE_COPY, + SCHEDULED_TASK_SEND_MODES, +} from "@t3tools/client-runtime/scheduled-task-send-mode"; import { tryCopyTextWithHaptic } from "../../lib/copyTextWithHaptic"; import { usePreparedConnection } from "../../state/session"; import { buildModelOptions } from "../../lib/modelOptions"; @@ -795,6 +799,7 @@ function TaskForm({ schedule, enabled: draft.enabled, threadId: draft.task?.threadId ?? null, + sendMode: draft.sendMode, workspaceStrategy: draft.workspace === "root" ? { type: "root" } @@ -932,6 +937,31 @@ function TaskForm({ /> + {draft.task?.threadId ? ( + + ({ + id: mode, + title: SCHEDULED_TASK_SEND_MODE_COPY[mode].label, + state: mode === draft.sendMode ? "on" : undefined, + })) + } + onSelect={(id) => { + const mode = SCHEDULED_TASK_SEND_MODES.find((candidate) => candidate === id); + if (mode) setDraft({ ...draft, sendMode: mode }); + }} + /> + + ) : null} + { schedule: { type: "webhook", signature: null }, projectId, threadId: null, + sendMode: "queue", workspaceStrategy: { type: "root" }, modelSelection: { instanceId: ProviderInstanceId.make("codex"), model: "gpt-5" }, runtimeMode: "approval-required", @@ -2747,6 +2748,7 @@ describe("OrchestratorMcpService provider resolution", () => { schedule: { type: "interval", everyMs: 3_600_000 }, projectId, threadId: null, + sendMode: "queue", workspaceStrategy: { type: "root" }, modelSelection: { instanceId: ProviderInstanceId.make("codex"), model: "gpt-5" }, runtimeMode: "approval-required", diff --git a/apps/server/src/persistence/Migrations.ts b/apps/server/src/persistence/Migrations.ts index 606770d1b6508..b7065c43a4333 100644 --- a/apps/server/src/persistence/Migrations.ts +++ b/apps/server/src/persistence/Migrations.ts @@ -83,6 +83,7 @@ import Migration0077 from "./Migrations/077_RepairBooleanSettledOverride.ts"; import Migration0078 from "./Migrations/078_DropOrchestrationEventCorrelationIndex.ts"; import Migration0079 from "./Migrations/079_DelegatedRecoveryIndexes.ts"; +import Migration0080 from "./Migrations/080_ScheduledTaskSendMode.ts"; const migrationEntries = [ [1, "OrchestrationEvents", Migration0001], @@ -165,6 +166,7 @@ const migrationEntries = [ [77, "RepairBooleanSettledOverride", Migration0077], [78, "DropOrchestrationEventCorrelationIndex", Migration0078], [79, "DelegatedRecoveryIndexes", Migration0079], + [80, "ScheduledTaskSendMode", Migration0080], ] as const; export const migrationManifest = migrationEntries.map(([id, name]) => [id, name] as const); diff --git a/apps/server/src/persistence/Migrations/080_ScheduledTaskSendMode.ts b/apps/server/src/persistence/Migrations/080_ScheduledTaskSendMode.ts new file mode 100644 index 0000000000000..de5a18489333d --- /dev/null +++ b/apps/server/src/persistence/Migrations/080_ScheduledTaskSendMode.ts @@ -0,0 +1,7 @@ +import * as Effect from "effect/Effect"; +import * as SqlClient from "effect/sql/SqlClient"; + +export default Effect.gen(function* () { + const sql = yield* SqlClient.SqlClient; + yield* sql`ALTER TABLE scheduled_tasks ADD COLUMN send_mode TEXT NOT NULL DEFAULT 'queue'`; +}); diff --git a/apps/server/src/persistence/reconcileV2PreviewMigration.test.ts b/apps/server/src/persistence/reconcileV2PreviewMigration.test.ts index 5f51c237fd763..65645310ef7ce 100644 --- a/apps/server/src/persistence/reconcileV2PreviewMigration.test.ts +++ b/apps/server/src/persistence/reconcileV2PreviewMigration.test.ts @@ -59,6 +59,7 @@ describe("V2 preview upgrade", () => { [77, "RepairBooleanSettledOverride"], [78, "DropOrchestrationEventCorrelationIndex"], [79, "DelegatedRecoveryIndexes"], + [80, "ScheduledTaskSendMode"], ]); assert.deepStrictEqual(yield* runMigrations(), []); assert.deepStrictEqual(yield* sql`SELECT * FROM orchestration_v2_legacy_imports`, imports); @@ -161,6 +162,7 @@ describe("V2 preview upgrade", () => { [77, "RepairBooleanSettledOverride"], [78, "DropOrchestrationEventCorrelationIndex"], [79, "DelegatedRecoveryIndexes"], + [80, "ScheduledTaskSendMode"], ]); }).pipe(Effect.provide(NodeSqliteClient.layer({ filename: ":memory:" }))), ); diff --git a/apps/server/src/scheduledTasks/ScheduledTaskService.test.ts b/apps/server/src/scheduledTasks/ScheduledTaskService.test.ts index c7b0e31f4f04d..cd397bd9503ae 100644 --- a/apps/server/src/scheduledTasks/ScheduledTaskService.test.ts +++ b/apps/server/src/scheduledTasks/ScheduledTaskService.test.ts @@ -8,6 +8,7 @@ import { ProviderInstanceId, ScheduledTaskError, ScheduledTaskId, + type ScheduledTaskUpsertInput, ThreadId, } from "@t3tools/contracts"; import * as DateTime from "effect/DateTime"; @@ -414,7 +415,9 @@ it.effect("a bound run follows its thread when promotion moves it mid-send", () Effect.gen(function* () { const scratch = ProjectId.make("project:scratch"); const promoted = ProjectId.make("project:promoted"); - const sends = yield* Ref.make>([]); + const sends = yield* Ref.make>( + [], + ); // Rebind lands first, then the promotion commits between the shell read and the retry. const shellProjects = yield* Ref.make([scratch, promoted]); const layer = ScheduledTaskService.layer.pipe( @@ -426,7 +429,7 @@ it.effect("a bound run follows its thread when promotion moves it mid-send", () sendToThread: (input) => Ref.updateAndGet(sends, (all) => [ ...all, - { projectId: input.projectId, commandId: input.commandId }, + { projectId: input.projectId, commandId: input.commandId, mode: input.mode }, ]).pipe( Effect.flatMap((all) => all.length < 3 @@ -458,7 +461,7 @@ it.effect("a bound run follows its thread when promotion moves it mid-send", () yield* Effect.gen(function* () { const service = yield* ScheduledTaskService.ScheduledTaskService; const threadId = ThreadId.make("thread:promoted-during-run"); - const { task } = yield* service.upsert({ + const input: ScheduledTaskUpsertInput = { id: ScheduledTaskId.make("task:promoted-during-run"), title: "Check in", prompt: "Check progress", @@ -470,7 +473,10 @@ it.effect("a bound run follows its thread when promotion moves it mid-send", () modelSelection: { instanceId: ProviderInstanceId.make("codex"), model: "gpt-5.4" }, runtimeMode: "full-access", interactionMode: "default", - }); + }; + yield* service.upsert({ ...input, sendMode: "auto" }); + // A save from a client or tool that predates the mode keeps it. + const { task } = yield* service.upsert(input); const result = yield* service.runNow({ id: task.id }); @@ -481,6 +487,7 @@ it.effect("a bound run follows its thread when promotion moves it mid-send", () [promoted, scratch, promoted], ); assert.equal(new Set(attempts.map((attempt) => attempt.commandId)).size, 1); + assert.deepEqual(new Set(attempts.map((attempt) => attempt.mode)), new Set(["auto"])); }).pipe(Effect.provide(layer)); }).pipe(Effect.provide(SqlitePersistence.layerMemory)), ); diff --git a/apps/server/src/scheduledTasks/ScheduledTaskService.ts b/apps/server/src/scheduledTasks/ScheduledTaskService.ts index 6d5e4484860fc..c1725b06d2c55 100644 --- a/apps/server/src/scheduledTasks/ScheduledTaskService.ts +++ b/apps/server/src/scheduledTasks/ScheduledTaskService.ts @@ -199,6 +199,7 @@ interface ScheduledTaskRow { readonly schedule_json: string; readonly project_id: string; readonly thread_id: string | null; + readonly send_mode: string; readonly workspace_strategy_json: string; readonly model_selection_json: string; readonly runtime_mode: string; @@ -372,6 +373,7 @@ const decodeRow = Effect.fnUntraced( schedule, projectId: row.project_id, threadId: row.thread_id, + sendMode: row.send_mode, workspaceStrategy, modelSelection, runtimeMode: row.runtime_mode, @@ -494,6 +496,7 @@ export const layer = Layer.effect( schedule_json, project_id, thread_id, + send_mode, workspace_strategy_json, model_selection_json, runtime_mode, @@ -535,6 +538,7 @@ export const layer = Layer.effect( schedule_json, project_id, thread_id, + send_mode, workspace_strategy_json, model_selection_json, runtime_mode, @@ -612,6 +616,7 @@ export const layer = Layer.effect( schedule_json, project_id, thread_id, + send_mode, workspace_strategy_json, model_selection_json, runtime_mode, @@ -636,6 +641,7 @@ export const layer = Layer.effect( ${JSON.stringify(task.schedule)}, ${task.projectId}, ${task.threadId}, + ${task.sendMode}, ${JSON.stringify(task.workspaceStrategy)}, ${JSON.stringify(task.modelSelection)}, ${task.runtimeMode}, @@ -661,6 +667,7 @@ export const layer = Layer.effect( schedule_json = excluded.schedule_json, project_id = excluded.project_id, thread_id = excluded.thread_id, + send_mode = excluded.send_mode, workspace_strategy_json = excluded.workspace_strategy_json, model_selection_json = excluded.model_selection_json, runtime_mode = excluded.runtime_mode, @@ -902,7 +909,7 @@ export const layer = Layer.effect( scheduledTaskId: active.id, text: prompt, attachments: [], - mode: "queue", + mode: active.sendMode, createdBy: active.createdBy, creationSource: active.creationSource, }) @@ -1328,6 +1335,8 @@ export const layer = Layer.effect( schedule, projectId: input.projectId, threadId: input.threadId ?? null, + // Omitted keeps the stored mode, so clients and tools that predate it keep it on re-save. + sendMode: input.sendMode ?? existingTask?.sendMode ?? "queue", workspaceStrategy, modelSelection: input.modelSelection, runtimeMode: input.runtimeMode, diff --git a/apps/web/src/components/settings/ScheduledTasksSettings.browser.test.tsx b/apps/web/src/components/settings/ScheduledTasksSettings.browser.test.tsx index ca8192dd894e3..85ce1104aaffb 100644 --- a/apps/web/src/components/settings/ScheduledTasksSettings.browser.test.tsx +++ b/apps/web/src/components/settings/ScheduledTasksSettings.browser.test.tsx @@ -26,6 +26,7 @@ const task: ScheduledTask = { schedule: { type: "interval", everyMs: 60_000 }, projectId, threadId: ThreadId.make("older-thread"), + sendMode: "queue", workspaceStrategy: { type: "root" }, modelSelection: { instanceId: ProviderInstanceId.make("codex"), model: "gpt-5.4" }, runtimeMode: "full-access", diff --git a/apps/web/src/components/settings/ScheduledTasksSettings.tsx b/apps/web/src/components/settings/ScheduledTasksSettings.tsx index 2b8b37d4790e3..7059b7e2f4717 100644 --- a/apps/web/src/components/settings/ScheduledTasksSettings.tsx +++ b/apps/web/src/components/settings/ScheduledTasksSettings.tsx @@ -20,6 +20,7 @@ import type { ScheduledTask, ScheduledTaskId, ScheduledTaskSchedule, + ScheduledTaskSendMode, ScheduledTaskUpsertInput, ScheduledTaskWebhookDeliverySummary, ThreadId, @@ -32,6 +33,10 @@ import { WEBHOOK_FILTER_OPS, type WebhookFilterDraft, } from "@t3tools/client-runtime/scheduled-task-webhook"; +import { + SCHEDULED_TASK_SEND_MODE_COPY, + SCHEDULED_TASK_SEND_MODES, +} from "@t3tools/client-runtime/scheduled-task-send-mode"; import { MAX_WEBHOOK_DELIVERY_AGE_MINUTES, MAX_WEBHOOK_FILTERS, @@ -149,6 +154,7 @@ const EMPTY_DRAFT: DraftState = { weekdays: new Set([1, 2, 3, 4, 5]), projectId: "", threadId: "", + sendMode: "queue", workspaceMode: "worktree", baseRef: "main", startFromOrigin: true, @@ -1356,6 +1362,7 @@ function ScheduledTaskEditorForm({ schedule, projectId: selectedProjectId as ProjectId, threadId: draft.threadId ? (draft.threadId as ThreadId) : null, + sendMode: draft.sendMode, workspaceStrategy, modelSelection, // Tasks have no mode picker; a mode above the administrator's ceiling would fail every run. @@ -1484,6 +1491,35 @@ function ScheduledTaskEditorForm({ threadId={draft.threadId ? (draft.threadId as ThreadId) : null} /> + {draft.threadId ? ( + + +

+ {SCHEDULED_TASK_SEND_MODE_COPY[draft.sendMode].description} +

+ + ) : null} +
Date: Thu, 5 Nov 2026 01:34:38 +0400 Subject: [PATCH 3/8] test(server): the steer retry test asserts its replayed delivery The stub projection now holds the accepted steer in its completed run, so the retry is checked to succeed as "steered" rather than only to reach dispatch. Co-Authored-By: Claude Opus 5.5 --- .../ThreadManagementService.test.ts | 42 ++++++++++++------- 1 file changed, 27 insertions(+), 15 deletions(-) diff --git a/apps/server/src/orchestration-v2/ThreadManagementService.test.ts b/apps/server/src/orchestration-v2/ThreadManagementService.test.ts index 756a1cb045904..aa421d72900c4 100644 --- a/apps/server/src/orchestration-v2/ThreadManagementService.test.ts +++ b/apps/server/src/orchestration-v2/ThreadManagementService.test.ts @@ -98,8 +98,22 @@ const layerDispatchChokepointTest = model: "test", }, }, - runs: [], + // The steer retry test's send, accepted into a turn that has since ended. + runs: [{ id: RunId.make("run:thread-management:steer-replay"), status: "completed" }], providerTurns: [], + messages: [ + { + id: MessageId.make("message:thread-management:steer-replay"), + runId: RunId.make("run:thread-management:steer-replay"), + }, + ], + turnItems: [ + { + type: "user_message", + messageId: MessageId.make("message:thread-management:steer-replay"), + inputIntent: "steer", + }, + ], } as unknown as OrchestrationV2ThreadProjection), dispatch: (command) => Effect.sync(() => { @@ -221,21 +235,19 @@ it.effect("a retried steer replays its accepted send after the steered turn has }); const threads = yield* ThreadManagementService.ThreadManagementService; - // The stub projection has no messages, so the send fails after dispatch; only the dispatch matters. - yield* Effect.exit( - threads.sendToThread({ - projectId, - commandId, - threadId, - messageId: MessageId.make("message:thread-management:steer-replay"), - text: "CI failed.", - attachments: [], - mode: "steer", - createdBy: "user", - creationSource: "server", - }), - ); + const result = yield* threads.sendToThread({ + projectId, + commandId, + threadId, + messageId: MessageId.make("message:thread-management:steer-replay"), + text: "CI failed.", + attachments: [], + mode: "steer", + createdBy: "user", + creationSource: "server", + }); + expect(result.delivery).toBe("steered"); expect(dispatched).toEqual(["message.dispatch"]); }).pipe(Effect.provide(layerTest)); }); From d85660253bb1639abdde3bed9a03f7ed6af03a2e Mon Sep 17 00:00:00 2001 From: 3xcalibur Date: Thu, 5 Nov 2026 23:55:02 +0400 Subject: [PATCH 4/8] fix(scheduled-tasks): Steer only and Restart copy matches composer steering The orchestrator can still queue or start a steer in edge cases, as it does for steering from the composer, so the copy no longer promises more. Co-Authored-By: Claude Opus 5.5 --- docs/user/project-settings.md | 7 +++---- packages/client-runtime/src/scheduledTaskSendMode.ts | 4 ++-- 2 files changed, 5 insertions(+), 6 deletions(-) diff --git a/docs/user/project-settings.md b/docs/user/project-settings.md index 07a3677b1581f..0196744c45444 100644 --- a/docs/user/project-settings.md +++ b/docs/user/project-settings.md @@ -100,10 +100,9 @@ A task whose runs continue one conversation has a **Delivery** setting for runs that arrive while the agent is still working there. **Queue**, the default, waits for the current turn to finish. **Steer or start** adds the prompt to the current turn when it can take new input, such as while the agent is not waiting on an -approval, and otherwise works like Queue. **Steer only** adds it to the current -turn, and **Restart** stops that turn and starts over with the prompt. Both need -a turn that can take new input: otherwise nothing is sent and the run is marked -failed. +approval, and otherwise works like Queue. **Steer only** and **Restart** act on +the current turn the way steering and restarting from the composer do. With no +turn to act on, nothing is sent and the run is marked failed. You can edit, pause, resume, run immediately, or delete a task from the list. Webhook tasks only run when their URL is called, so they can't be run diff --git a/packages/client-runtime/src/scheduledTaskSendMode.ts b/packages/client-runtime/src/scheduledTaskSendMode.ts index 19c56d78a19b6..e72d59f64f9f2 100644 --- a/packages/client-runtime/src/scheduledTaskSendMode.ts +++ b/packages/client-runtime/src/scheduledTaskSendMode.ts @@ -18,11 +18,11 @@ export const SCHEDULED_TASK_SEND_MODE_COPY: Record< steer: { label: "Steer only", description: - "Steers the current turn. Without a turn that can take new input, nothing is sent and the run fails.", + "Steers the current turn, like steering from the composer. With no turn to steer, nothing is sent and the run fails.", }, restart: { label: "Restart", description: - "Stops the current turn and starts over with this prompt. Without a turn that can take new input, nothing is sent and the run fails.", + "Stops the current turn and starts over with this prompt, like restarting from the composer. With no turn to restart, nothing is sent and the run fails.", }, }; From 49f79265f1f953d27e5c004315dc955c17410733 Mon Sep 17 00:00:00 2001 From: 3xcalibur Date: Fri, 6 Nov 2026 00:03:25 +0400 Subject: [PATCH 5/8] fix(scheduled-tasks): Steer only and Restart copy notes the first run starts the thread Co-Authored-By: Claude Opus 5.5 --- docs/user/project-settings.md | 3 ++- packages/client-runtime/src/scheduledTaskSendMode.ts | 4 ++-- 2 files changed, 4 insertions(+), 3 deletions(-) diff --git a/docs/user/project-settings.md b/docs/user/project-settings.md index 0196744c45444..a2a93c7072b6c 100644 --- a/docs/user/project-settings.md +++ b/docs/user/project-settings.md @@ -102,7 +102,8 @@ current turn to finish. **Steer or start** adds the prompt to the current turn when it can take new input, such as while the agent is not waiting on an approval, and otherwise works like Queue. **Steer only** and **Restart** act on the current turn the way steering and restarting from the composer do. With no -turn to act on, nothing is sent and the run is marked failed. +turn to act on, nothing is sent and the run is marked failed. The first run +always starts the thread. You can edit, pause, resume, run immediately, or delete a task from the list. Webhook tasks only run when their URL is called, so they can't be run diff --git a/packages/client-runtime/src/scheduledTaskSendMode.ts b/packages/client-runtime/src/scheduledTaskSendMode.ts index e72d59f64f9f2..f6ecde8a31daa 100644 --- a/packages/client-runtime/src/scheduledTaskSendMode.ts +++ b/packages/client-runtime/src/scheduledTaskSendMode.ts @@ -18,11 +18,11 @@ export const SCHEDULED_TASK_SEND_MODE_COPY: Record< steer: { label: "Steer only", description: - "Steers the current turn, like steering from the composer. With no turn to steer, nothing is sent and the run fails.", + "Steers the current turn, like steering from the composer. With no turn to steer, nothing is sent and the run fails. The first run starts the thread.", }, restart: { label: "Restart", description: - "Stops the current turn and starts over with this prompt, like restarting from the composer. With no turn to restart, nothing is sent and the run fails.", + "Stops the current turn and starts over with this prompt, like restarting from the composer. With no turn to restart, nothing is sent and the run fails. The first run starts the thread.", }, }; From 01cb5af0c64031442eaa4226458b4a1776d3001e Mon Sep 17 00:00:00 2001 From: 3xcalibur Date: Sun, 15 Nov 2026 01:48:09 +0400 Subject: [PATCH 6/8] fix(mobile): delivery mode picker uses the updater-only setDraft from main Co-Authored-By: Claude Opus 5.5 --- .../settings/SettingsScheduledTasksRouteScreen.test.tsx | 1 + .../src/features/settings/SettingsScheduledTasksRouteScreen.tsx | 2 +- 2 files changed, 2 insertions(+), 1 deletion(-) diff --git a/apps/mobile/src/features/settings/SettingsScheduledTasksRouteScreen.test.tsx b/apps/mobile/src/features/settings/SettingsScheduledTasksRouteScreen.test.tsx index b92c1ec725b47..b736797515814 100644 --- a/apps/mobile/src/features/settings/SettingsScheduledTasksRouteScreen.test.tsx +++ b/apps/mobile/src/features/settings/SettingsScheduledTasksRouteScreen.test.tsx @@ -158,6 +158,7 @@ beforeEach(() => { schedule: { type: "fixed_time", timeOfDay: "05:00" }, projectId: ProjectId.make("project"), threadId: null, + sendMode: "queue", workspaceStrategy: { type: "root" }, modelSelection: { instanceId: ProviderInstanceId.make("codex"), diff --git a/apps/mobile/src/features/settings/SettingsScheduledTasksRouteScreen.tsx b/apps/mobile/src/features/settings/SettingsScheduledTasksRouteScreen.tsx index 05f8da2ae3e91..c26685223f924 100644 --- a/apps/mobile/src/features/settings/SettingsScheduledTasksRouteScreen.tsx +++ b/apps/mobile/src/features/settings/SettingsScheduledTasksRouteScreen.tsx @@ -1024,7 +1024,7 @@ function TaskForm({ } onSelect={(id) => { const mode = SCHEDULED_TASK_SEND_MODES.find((candidate) => candidate === id); - if (mode) setDraft({ ...draft, sendMode: mode }); + if (mode) setDraft((current) => ({ ...current, sendMode: mode })); }} /> From 401f0514ae23cfcdbb870221630aceb73efe4cd2 Mon Sep 17 00:00:00 2001 From: maria-rcks Date: Sun, 15 Nov 2026 03:12:41 +0000 Subject: [PATCH 7/8] fix(scheduled-tasks): preserve delivery mode and track steered failures --- .../ScheduledTaskService.test.ts | 34 +++++++++++++++---- .../scheduledTasks/ScheduledTaskService.ts | 14 ++++++-- 2 files changed, 39 insertions(+), 9 deletions(-) diff --git a/apps/server/src/scheduledTasks/ScheduledTaskService.test.ts b/apps/server/src/scheduledTasks/ScheduledTaskService.test.ts index 8c0b9a3b65153..72e7c1cd25e29 100644 --- a/apps/server/src/scheduledTasks/ScheduledTaskService.test.ts +++ b/apps/server/src/scheduledTasks/ScheduledTaskService.test.ts @@ -420,10 +420,12 @@ it.effect("thread cleanup notifies subscribers and a finishing run cannot recrea ); it.effect.each([ - { status: "failed", error: "Weekly usage limit reached" }, - { status: "interrupted", error: "The run was stopped." }, - { status: "cancelled", error: "The run was cancelled." }, -] as const)("a run that ends $status after its message was accepted records it", (outcome) => + { status: "failed", error: "Weekly usage limit reached", mode: "queue" }, + { status: "failed", error: "Weekly usage limit reached", mode: "steer" }, + { status: "failed", error: "Weekly usage limit reached", mode: "auto" }, + { status: "interrupted", error: "The run was stopped.", mode: "queue" }, + { status: "cancelled", error: "The run was cancelled.", mode: "queue" }, +] as const)("a $mode run that ends $status after its message was accepted records it", (outcome) => Effect.gen(function* () { const projectId = ProjectId.make("project:test"); const ended = yield* Deferred.make(); @@ -444,7 +446,10 @@ it.effect.each([ ({ runs: [ { id: "run:1", status: "completed", userMessageId: "message:earlier" }, - { ...run, userMessageId }, + { + ...run, + userMessageId: outcome.mode === "queue" ? userMessageId : "message:active", + }, ], messages: [{ id: userMessageId, runId: "run:2" }], turnItems: [ @@ -482,6 +487,7 @@ it.effect.each([ const { task } = yield* service.upsert({ id: ScheduledTaskId.make("task:quota"), title: "Quota bookkeeping", + sendMode: outcome.mode, prompt: "Update the notes", enabled: true, schedule: { type: "interval", everyMs: 60_000 }, @@ -511,6 +517,8 @@ it.effect("a bound run follows its thread when promotion moves it mid-send", () Effect.gen(function* () { const scratch = ProjectId.make("project:scratch"); const promoted = ProjectId.make("project:promoted"); + const saving = yield* Deferred.make(); + const resumeSave = yield* Deferred.make(); const sends = yield* Ref.make>( [], ); @@ -545,7 +553,11 @@ it.effect("a bound run follows its thread when promotion moves it mid-send", () Layer.mock(SecretRequests.SecretRequests)({}), Layer.mock(ManagedProjectFolders.ManagedProjectFolders)({ namedProjectsRoot: "/projects", - isScratchProject: () => Effect.succeed(false), + isScratchProject: () => + Deferred.succeed(saving, undefined).pipe( + Effect.andThen(Deferred.await(resumeSave)), + Effect.as(true), + ), }), NodeCrypto.layer, Scheduler.layer, @@ -582,6 +594,16 @@ it.effect("a bound run follows its thread when promotion moves it mid-send", () ); assert.equal(new Set(attempts.map((attempt) => attempt.commandId)).size, 1); assert.deepEqual(new Set(attempts.map((attempt) => attempt.mode)), new Set(["auto"])); + const staleSave = yield* service + .upsert({ + ...input, + workspaceStrategy: { type: "worktree", branch: "task", baseRef: "main" }, + }) + .pipe(Effect.forkChild); + yield* Deferred.await(saving); + yield* service.upsert({ ...input, sendMode: "restart" }); + yield* Deferred.succeed(resumeSave, undefined); + assert.equal((yield* Fiber.join(staleSave)).task.sendMode, "restart"); }).pipe(Effect.provide(layer)); }).pipe(Effect.provide(SqlitePersistence.layerMemory)), ); diff --git a/apps/server/src/scheduledTasks/ScheduledTaskService.ts b/apps/server/src/scheduledTasks/ScheduledTaskService.ts index d435467380374..e6d058fef041d 100644 --- a/apps/server/src/scheduledTasks/ScheduledTaskService.ts +++ b/apps/server/src/scheduledTasks/ScheduledTaskService.ts @@ -627,6 +627,7 @@ export const layer = Layer.effect( const saveTask = ( task: ScheduledTask, requireExisting: boolean, + sendMode: ScheduledTaskUpsertInput["sendMode"], /** The schedule this save read, when it left the schedule unchanged. */ keptSchedule: ScheduledTask["schedule"] | null, webhook: { @@ -704,7 +705,7 @@ export const layer = Layer.effect( ELSE COALESCE(excluded.thread_id, scheduled_tasks.thread_id) END, thread_mode = excluded.thread_mode, - send_mode = excluded.send_mode, + send_mode = COALESCE(${sendMode ?? null}, scheduled_tasks.send_mode), workspace_strategy_json = excluded.workspace_strategy_json, model_selection_json = excluded.model_selection_json, runtime_mode = excluded.runtime_mode, @@ -890,8 +891,14 @@ export const layer = Layer.effect( const shell = yield* threadManagement.getThreadShell(threadId); if (shell === null) return; const thread = { projectId: shell.projectId, threadId }; - const { runs } = yield* threadManagement.getProjectThreadRecords(thread, ["runs"]); - const dispatched = runs.find((run) => run.userMessageId === messageId); + const { runs, messages: dispatchedMessages } = + yield* threadManagement.getProjectThreadRecords(thread, ["runs", "messages"], { + messageIds: [messageId], + }); + const messageRunId = dispatchedMessages.find((message) => message.id === messageId)?.runId; + const dispatched = + runs.find((run) => run.userMessageId === messageId) ?? + runs.find((run) => run.id === messageRunId); if (dispatched === undefined) return; const { run } = yield* threadManagement.waitForThread({ ...thread, @@ -1561,6 +1568,7 @@ export const layer = Layer.effect( yield* saveTask( task, input.requireExisting === true, + input.sendMode, scheduleUnchanged ? existingTask.schedule : null, webhook, ); From 9bb951637c0e41490219f55e9dc9940eedddcbde Mon Sep 17 00:00:00 2001 From: maria-rcks Date: Sun, 15 Nov 2026 03:26:14 +0000 Subject: [PATCH 8/8] fix(scheduled-tasks): follow delivery fallback runs --- .../094_TurnItemMessageOrdinalIndex.test.ts | 2 +- .../ScheduledTaskService.test.ts | 17 ++++-- .../scheduledTasks/ScheduledTaskService.ts | 61 ++++++++++--------- 3 files changed, 46 insertions(+), 34 deletions(-) diff --git a/apps/server/src/persistence/Migrations/094_TurnItemMessageOrdinalIndex.test.ts b/apps/server/src/persistence/Migrations/094_TurnItemMessageOrdinalIndex.test.ts index 1a8ea4ecfed43..5d76b7d1a95b9 100644 --- a/apps/server/src/persistence/Migrations/094_TurnItemMessageOrdinalIndex.test.ts +++ b/apps/server/src/persistence/Migrations/094_TurnItemMessageOrdinalIndex.test.ts @@ -43,7 +43,7 @@ it.effect("migrates over a message whose payload is not JSON, and takes one afte yield* insertTurnItem("ok-1", "thread-a", 2, "user_message", '{"messageId":"m1"}'); assert.deepEqual( - (yield* runMigrations()).map(([id]) => id), + (yield* runMigrations({ toMigrationInclusive: 94 })).map(([id]) => id), [94], ); yield* insertTurnItem("bad-2", "thread-a", 3, "user_message", "{not json"); diff --git a/apps/server/src/scheduledTasks/ScheduledTaskService.test.ts b/apps/server/src/scheduledTasks/ScheduledTaskService.test.ts index 72e7c1cd25e29..79c602b84a656 100644 --- a/apps/server/src/scheduledTasks/ScheduledTaskService.test.ts +++ b/apps/server/src/scheduledTasks/ScheduledTaskService.test.ts @@ -423,6 +423,7 @@ it.effect.each([ { status: "failed", error: "Weekly usage limit reached", mode: "queue" }, { status: "failed", error: "Weekly usage limit reached", mode: "steer" }, { status: "failed", error: "Weekly usage limit reached", mode: "auto" }, + { status: "interrupted", error: "The run was stopped.", mode: "auto", followUp: true }, { status: "interrupted", error: "The run was stopped.", mode: "queue" }, { status: "cancelled", error: "The run was cancelled.", mode: "queue" }, ] as const)("a $mode run that ends $status after its message was accepted records it", (outcome) => @@ -430,6 +431,8 @@ it.effect.each([ const projectId = ProjectId.make("project:test"); const ended = yield* Deferred.make(); const sent = yield* Ref.make(""); + const messageRun = yield* Ref.make("run:2"); + const followUp = "followUp" in outcome; const run = { id: "run:2", status: outcome.status, rootNodeId: "node:root" }; const layer = ScheduledTaskService.layer.pipe( Layer.provide( @@ -441,8 +444,8 @@ it.effect.each([ getThreadShell: () => Effect.succeed(threadShell(projectId)), getProjectThreadRecords: () => Effect.map( - Ref.get(sent), - (userMessageId) => + Effect.all([Ref.get(sent), Ref.get(messageRun)]), + ([userMessageId, runId]) => ({ runs: [ { id: "run:1", status: "completed", userMessageId: "message:earlier" }, @@ -451,7 +454,7 @@ it.effect.each([ userMessageId: outcome.mode === "queue" ? userMessageId : "message:active", }, ], - messages: [{ id: userMessageId, runId: "run:2" }], + messages: [{ id: userMessageId, runId }], turnItems: [ { type: "error", @@ -465,9 +468,15 @@ it.effect.each([ ), waitForThread: (input) => Deferred.await(ended).pipe( + Effect.andThen(Ref.set(messageRun, followUp ? "run:3" : "run:2")), Effect.as({ threadId: input.threadId, - run: { ...run, id: input.runId, userMessageId: "" }, + run: { + ...run, + id: input.runId, + userMessageId: "", + status: followUp && input.runId === "run:2" ? "completed" : outcome.status, + }, timedOut: false, } as never), ), diff --git a/apps/server/src/scheduledTasks/ScheduledTaskService.ts b/apps/server/src/scheduledTasks/ScheduledTaskService.ts index e6d058fef041d..fa583d127e162 100644 --- a/apps/server/src/scheduledTasks/ScheduledTaskService.ts +++ b/apps/server/src/scheduledTasks/ScheduledTaskService.ts @@ -900,36 +900,39 @@ export const layer = Layer.effect( runs.find((run) => run.userMessageId === messageId) ?? runs.find((run) => run.id === messageRunId); if (dispatched === undefined) return; - const { run } = yield* threadManagement.waitForThread({ - ...thread, - runId: dispatched.id, - timeoutMs: RUN_OUTCOME_WAIT_MS, - }); - if ( - run === null || - run.status === "completed" || - !ThreadManagementService.isTerminalRunStatus(run.status) - ) - return; - const { turnItems, messages } = yield* threadManagement.getProjectThreadRecords(thread, [ - "turnItems", - "messages", - ]); - // A queued message promoted to a steer cancels its run and joins the active one. - if ( - run.status === "cancelled" && - messages.find((message) => message.id === messageId)?.runId !== run.id - ) + let runId = dispatched.id; + while (true) { + const { run } = yield* threadManagement.waitForThread({ + ...thread, + runId, + timeoutMs: RUN_OUTCOME_WAIT_MS, + }); + if (run === null || !ThreadManagementService.isTerminalRunStatus(run.status)) return; + const { messages } = yield* threadManagement.getProjectThreadRecords(thread, ["messages"], { + messageIds: [messageId], + }); + const currentRunId = messages.find((message) => message.id === messageId)?.runId; + // A declined steer or a promoted queued message can move into another run. + if (currentRunId != null && currentRunId !== runId) { + runId = currentRunId; + continue; + } + if (run.status === "completed" || (run.status === "cancelled" && currentRunId == null)) + return; + const { turnItems } = yield* threadManagement.getProjectThreadRecords(thread, [ + "turnItems", + ]); + const error = + latestRootProviderFailure(run, turnItems)?.message ?? UNSUCCESSFUL_RUN_ERRORS[run.status]; + const updated = yield* sql` + UPDATE scheduled_tasks + SET last_run_status = 'failed', last_run_error = ${error} + WHERE task_id = ${taskId} AND last_run_at = ${startedAtIso} AND last_run_status = 'succeeded' + RETURNING task_id + `; + if (updated.length > 0) yield* notifyChanged; return; - const error = - latestRootProviderFailure(run, turnItems)?.message ?? UNSUCCESSFUL_RUN_ERRORS[run.status]; - const updated = yield* sql` - UPDATE scheduled_tasks - SET last_run_status = 'failed', last_run_error = ${error} - WHERE task_id = ${taskId} AND last_run_at = ${startedAtIso} AND last_run_status = 'succeeded' - RETURNING task_id - `; - if (updated.length > 0) yield* notifyChanged; + } }); const runTask = Effect.fn("ScheduledTaskService.runTask")(function* (