feat(scheduled-tasks): bound tasks can steer, steer-or-start, or restart instead of queueing

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"),

apps/mobile/src/features/settings/SettingsScheduledTasksRouteScreen.tsx

@@ -58,6 +58,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";
@@ -810,6 +814,7 @@ function TaskForm({
       enabled: draft.enabled,
       threadId: draft.threadId,
       ...(offersThreadModes ? { threadMode: draft.threadMode } : {}),
+      ...(draft.sendMode === draft.task?.sendMode ? {} : { sendMode: draft.sendMode }),
       workspaceStrategy:
         draft.workspace === "root"
           ? { type: "root" }
@@ -999,6 +1004,32 @@ function TaskForm({
           ) : null}
         </SettingsSection>
 
+        {draft.threadMode === "continue" &&
+        config?.environment.capabilities.scheduledTaskSendMode === true ? (
+          <SettingsSection
+            title="Thread"
+            footer={SCHEDULED_TASK_SEND_MODE_COPY[draft.sendMode].description}
+          >
+            <SelectRow
+              label="Delivery"
+              value={SCHEDULED_TASK_SEND_MODE_COPY[draft.sendMode].label}
+              actions={
+                saving || dictationPending
+                  ? []
+                  : SCHEDULED_TASK_SEND_MODES.map((mode) => ({
+                      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((current) => ({ ...current, sendMode: mode }));
+              }}
+            />
+          </SettingsSection>
+        ) : null}
+
         <SettingsSection title="Workspace">
           <SelectRow
             label="Run in"

apps/mobile/src/features/settings/scheduledTaskDraft.ts

@@ -6,6 +6,7 @@ import type {
   RuntimeMode,
   ScheduledTask,
   ScheduledTaskThreadMode,
+  ScheduledTaskSendMode,
   ScheduledTaskUpsertSchedule,
   ScheduledTaskWebhookSignature,
   ThreadId,
@@ -214,6 +215,7 @@ export type ScheduledTaskDraft = {
   readonly enabled: boolean;
   readonly startFromOrigin: boolean;
   readonly runtimeMode: RuntimeMode;
+  readonly sendMode: ScheduledTaskSendMode;
 };
 
 function draftSignature(draft: ScheduledTaskDraft): string {
@@ -244,6 +246,7 @@ function draftSignature(draft: ScheduledTaskDraft): string {
     draft.enabled,
     draft.startFromOrigin,
     draft.runtimeMode,
+    draft.sendMode,
   ]);
 }
 
@@ -306,6 +309,7 @@ export function createDraft(
     enabled: true,
     startFromOrigin: true,
     runtimeMode: "full-access",
+    sendMode: "queue",
   };
 }
 
@@ -332,5 +336,6 @@ export function editDraft(task: ScheduledTask): ScheduledTaskDraft {
         ? (task.workspaceStrategy.startFromOrigin ?? false)
         : true,
     runtimeMode: task.runtimeMode,
+    sendMode: task.sendMode,
   };
 }

apps/server/src/environment/ServerEnvironment.ts

@@ -279,6 +279,7 @@ export const make = Effect.gen(function* () {
       oneTimeSchedules: true,
       freshContext: true,
       scheduledTaskThreadModes: true,
+      scheduledTaskSendMode: true,
       npmReleaseLookup: true,
       voiceTranscription: isSpeechPlatformSupported(hostPlatform, hostArchitecture, hostLibc),
       cloudVoiceTranscription: true,

apps/server/src/mcp/OrchestratorMcpService.test.ts

@@ -3387,6 +3387,7 @@ describe("OrchestratorMcpService provider resolution", () => {
       schedule: { type: "webhook", signature: null },
       projectId,
       threadId: null,
+      sendMode: "queue",
       workspaceStrategy: { type: "root" },
       modelSelection: { instanceId: ProviderInstanceId.make("codex"), model: "gpt-5" },
       runtimeMode: "approval-required",
@@ -3624,6 +3625,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",

apps/server/src/orchestration-v2/ThreadManagementService.test.ts

@@ -105,11 +105,28 @@ const layerDispatchChokepointTest =
                 id: ThreadId.make("thread:thread-management:chokepoint"),
                 projectId: ProjectId.make("project:thread-management:chokepoint"),
                 deletedAt: null,
+                archivedAt: null,
                 modelSelection: {
                   instanceId: ProviderInstanceId.make("codex_work"),
                   model: "test",
                 },
               },
+              // 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(() => {
@@ -211,6 +228,43 @@ it.effect(
   },
 );
 
+it.effect("a retried steer replays its accepted send after the steered turn has ended", () => {
+  const projectId = ProjectId.make("project:thread-management:chokepoint");
+  const threadId = ThreadId.make("thread:thread-management:chokepoint");
+  const commandId = CommandId.make("command:thread-management:steer-replay");
+  const dispatched: Array<string> = [];
+  const layerTest = layerDispatchChokepointTest(ServerSettings.layerTest())(dispatched);
+
+  return Effect.gen(function* () {
+    const receipts = yield* CommandReceiptStore.CommandReceiptStoreV2;
+    yield* receipts.insertIfAbsent({
+      commandId,
+      threadId,
+      commandType: "message.dispatch",
+      acceptedAt: yield* DateTime.now,
+      resultSequence: 1,
+      status: "accepted",
+      error: null,
+    });
+
+    const threads = yield* ThreadManagementService.ThreadManagementService;
+    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));
+});
+
 it.effect.each([
   { label: "fails a send that starts a turn", running: false },
   { label: "still steers a running turn", running: true },

apps/server/src/orchestration-v2/ThreadManagementService.ts

@@ -884,16 +884,20 @@ const make = Effect.gen(function* () {
       { readonly type: "message.dispatch" }
     >["dispatchMode"];
     if (mode === "steer" || mode === "restart") {
-      if (steerableRun === undefined) {
+      if (steerableRun === undefined && !(yield* hasReceipt(input.commandId))) {
         return yield* new ThreadManagementNoSteerableRunError({
           threadId: input.threadId,
           mode,
         });
       }
-      dispatchMode = {
-        type: mode === "steer" ? "steer_active" : "restart_active",
-        targetRunId: steerableRun.id,
-      };
+      // A retry of an accepted send replays its receipt, even after the steered turn has ended.
+      dispatchMode =
+        steerableRun === undefined
+          ? { type: "queue_after_active" }
+          : {
+              type: mode === "steer" ? "steer_active" : "restart_active",
+              targetRunId: steerableRun.id,
+            };
     } else if (mode === "auto" && steerableRun !== undefined) {
       dispatchMode = { type: "steer_active", targetRunId: steerableRun.id };
     } else {

apps/server/src/persistence/Migrations.ts

@@ -103,6 +103,7 @@ import Migration0092 from "./Migrations/092_QueuedTurnInputIndexJsonValid.ts";
 import Migration0093 from "./Migrations/093_ProviderTurnAttemptIndex.ts";
 import Migration0094 from "./Migrations/094_TurnItemMessageOrdinalIndex.ts";
 import Migration0095 from "./Migrations/095_TurnItemInlineWorkOutput.ts";
+import Migration0096 from "./Migrations/096_ScheduledTaskSendMode.ts";
 
 const migrationEntries = [
   [1, "OrchestrationEvents", Migration0001],
@@ -201,6 +202,7 @@ const migrationEntries = [
   [93, "ProviderTurnAttemptIndex", Migration0093],
   [94, "TurnItemMessageOrdinalIndex", Migration0094],
   [95, "TurnItemInlineWorkOutput", Migration0095],
+  [96, "ScheduledTaskSendMode", Migration0096],
 ] as const;
 
 export const migrationManifest = migrationEntries.map(([id, name]) => [id, name] as const);

apps/server/src/persistence/Migrations/096_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'`;
+});

apps/server/src/persistence/reconcileV2PreviewMigration.test.ts

@@ -75,6 +75,7 @@ describe("V2 preview upgrade", () => {
         [93, "ProviderTurnAttemptIndex"],
         [94, "TurnItemMessageOrdinalIndex"],
         [95, "TurnItemInlineWorkOutput"],
+        [96, "ScheduledTaskSendMode"],
       ]);
       assert.deepStrictEqual(yield* runMigrations(), []);
       assert.deepStrictEqual(yield* sql`SELECT * FROM orchestration_v2_legacy_imports`, imports);
@@ -193,6 +194,7 @@ describe("V2 preview upgrade", () => {
         [93, "ProviderTurnAttemptIndex"],
         [94, "TurnItemMessageOrdinalIndex"],
         [95, "TurnItemInlineWorkOutput"],
+        [96, "ScheduledTaskSendMode"],
       ]);
     }).pipe(Effect.provide(NodeSqliteClient.layer({ filename: ":memory:" }))),
   );