From b6766fc341d60edeefbc0a24db7d0458b380f61e Mon Sep 17 00:00:00 2001 From: Julius Marminge Date: Thu, 12 Nov 2026 06:42:28 +0000 Subject: [PATCH 1/5] test(server): a missed fixed-time slot is not shown on its scheduled task Refs #30844 AI-assisted; verified with the tests and tools listed above. --- .../ScheduledTaskService.schedule.test.ts | 73 +++++++++++++++++++ 1 file changed, 73 insertions(+) diff --git a/apps/server/src/scheduledTasks/ScheduledTaskService.schedule.test.ts b/apps/server/src/scheduledTasks/ScheduledTaskService.schedule.test.ts index 537d071853e34..15a48147f28d4 100644 --- a/apps/server/src/scheduledTasks/ScheduledTaskService.schedule.test.ts +++ b/apps/server/src/scheduledTasks/ScheduledTaskService.schedule.test.ts @@ -392,3 +392,76 @@ it.effect( }).pipe(Effect.provide(ScheduledTaskService.layer.pipe(Layer.provide(layerDependencies)))); }).pipe(Effect.provide(SqlitePersistence.layerMemory)), ); + +it.effect("a fixed-time slot missed while the server was off shows on the task", () => + Effect.gen(function* () { + const at = (hour: number, minute = 0, day = 1) => + DateTime.makeZonedUnsafe( + { year: 2026, month: 7, day, hour, minute, second: 0, millisecond: 0 }, + { timeZone: DateTime.zoneMakeLocal(), adjustForTimeZone: true }, + ); + yield* TestClock.setTime(DateTime.toEpochMillis(at(8))); + const launches = yield* Ref.make(0); + const layerDependencies = Layer.mergeAll( + NodeCrypto.layer, + Scheduler.layer, + ManagedPolicy.layerTest(), + Layer.mock(ThreadLaunchService.ThreadLaunchService)({ + launch: () => + Ref.update(launches, (count) => count + 1).pipe( + Effect.as({ threadId: "thread-missed-slot", resumed: false } as never), + ), + }), + Layer.mock(ManagedProjectFolders.ManagedProjectFolders)({ + namedProjectsRoot: "/projects", + isScratchProject: () => Effect.succeed(false), + }), + Layer.mock(ThreadManagementService.ThreadManagementService)({}), + Layer.mock(SecretRequests.SecretRequests)({}), + ); + const withServer = (effect: Effect.Effect) => + effect.pipe( + Effect.provide(ScheduledTaskService.layer.pipe(Layer.provide(layerDependencies))), + ); + + const taskId = yield* withServer( + Effect.gen(function* () { + const service = yield* ScheduledTaskService.ScheduledTaskService; + const { task } = yield* service.upsert( + yield* decodeUpsertInput({ + title: "Morning bookkeeping", + prompt: "Update yesterday's notes.", + enabled: true, + schedule: { type: "fixed_time", timeOfDay: "09:00" }, + projectId: "project-missed-slot", + workspaceStrategy: { type: "root" }, + modelSelection: { instanceId: "codex", model: "gpt-5.4" }, + runtimeMode: "full-access", + interactionMode: "default", + }), + ); + yield* service.runNow({ id: task.id }); + return task.id; + }), + ); + + // The server is off across the 09:00 slot and comes back at 09:30. + yield* TestClock.setTime(DateTime.toEpochMillis(at(9, 30))); + const [listed] = yield* withServer( + Effect.gen(function* () { + const service = yield* ScheduledTaskService.ScheduledTaskService; + yield* TestClock.adjust("6 seconds"); + return (yield* service.list()).tasks; + }), + ); + + expect(listed?.id).toBe(taskId); + expect(yield* Ref.get(launches)).toBe(1); + expect(listed?.nextRunAt).toBe(DateTime.formatIso(DateTime.toUtc(at(9, 0, 2)))); + // The skipped slot must not read as the earlier manual run's clean success. + expect({ status: listed?.lastRunStatus, error: listed?.lastRunError }).not.toEqual({ + status: "succeeded", + error: null, + }); + }).pipe(Effect.provide(SqlitePersistence.layerMemory)), +); From 6ab3277bc3ab7fcb9d928673d232680da42818f3 Mon Sep 17 00:00:00 2001 From: Julius Marminge Date: Thu, 12 Nov 2026 06:53:21 +0000 Subject: [PATCH 2/5] fix(server): a missed fixed-time slot shows as failed on its scheduled task Fixes pingdotgg/t3code#30844 AI-assisted; verified with the tests and tools listed above. --- apps/server/src/scheduledTasks/ScheduledTaskService.ts | 9 ++++++++- 1 file changed, 8 insertions(+), 1 deletion(-) diff --git a/apps/server/src/scheduledTasks/ScheduledTaskService.ts b/apps/server/src/scheduledTasks/ScheduledTaskService.ts index e56f804286ef7..146ee587fafdc 100644 --- a/apps/server/src/scheduledTasks/ScheduledTaskService.ts +++ b/apps/server/src/scheduledTasks/ScheduledTaskService.ts @@ -1146,6 +1146,7 @@ export const layer = Layer.effect( task: ScheduledTask, now: DateTime.DateTime, reason = "missed", + runError?: string, ) { const next = nextRunAt(task, now); yield* Effect.logInfo("Skipping schedule task run", { @@ -1158,6 +1159,7 @@ export const layer = Layer.effect( UPDATE scheduled_tasks SET next_run_at = ${next}, updated_at = ${iso(now)} + ${runError === undefined ? sql`` : sql`, last_run_status = 'failed', last_run_error = ${runError}`} WHERE task_id = ${task.id} `.pipe( Effect.mapError((cause) => @@ -1202,7 +1204,12 @@ export const layer = Layer.effect( ({ task, dueAt }) => Effect.suspend(() => isMissedFixedTimeRun(task.schedule, dueAt, now) - ? rescheduleMissedRun(task, now) + ? rescheduleMissedRun( + task, + now, + "missed", + "Missed a scheduled run because the server was not running at that time.", + ) : runTask(task, "scheduled"), ).pipe( Effect.catch((cause) => From 922020b6cb3303d9ea207356a684ecc277db26ec Mon Sep 17 00:00:00 2001 From: Julius Marminge Date: Thu, 12 Nov 2026 07:18:28 +0000 Subject: [PATCH 3/5] fix(server): a scheduled run the provider fails afterwards shows as failed A run was recorded as succeeded once its message was accepted. Follow the dispatched run and, when it ends failed, record the provider's error on the task. Fixes pingdotgg/t3code#30844 AI-assisted; verified with the tests and tools listed above. --- .../ScheduledTaskService.test.ts | 84 +++++++++++++++++++ .../scheduledTasks/ScheduledTaskService.ts | 47 ++++++++++- 2 files changed, 129 insertions(+), 2 deletions(-) diff --git a/apps/server/src/scheduledTasks/ScheduledTaskService.test.ts b/apps/server/src/scheduledTasks/ScheduledTaskService.test.ts index cbfea6b4e3135..03cb9aec20407 100644 --- a/apps/server/src/scheduledTasks/ScheduledTaskService.test.ts +++ b/apps/server/src/scheduledTasks/ScheduledTaskService.test.ts @@ -17,6 +17,7 @@ import * as Effect from "effect/Effect"; import * as Exit from "effect/Exit"; import * as Layer from "effect/Layer"; import * as Logger from "effect/Logger"; +import * as Option from "effect/Option"; import * as SqlClient from "effect/sql/SqlClient"; import * as Fiber from "effect/Fiber"; @@ -418,6 +419,89 @@ it.effect("thread cleanup notifies subscribers and a finishing run cannot recrea }).pipe(Effect.provide(SqlitePersistence.layerMemory)), ); +it.effect("a run the provider fails after accepting its message records the failure", () => + Effect.gen(function* () { + const projectId = ProjectId.make("project:test"); + const ended = yield* Deferred.make(); + const sent = yield* Ref.make(""); + const run = { id: "run:2", status: "failed", rootNodeId: "node:root" }; + const layer = ScheduledTaskService.layer.pipe( + Layer.provide( + Layer.mergeAll( + ManagedPolicy.layerTest(), + Layer.mock(ThreadLaunchService.ThreadLaunchService)({}), + Layer.mock(ThreadManagementService.ThreadManagementService)({ + sendToThread: (input) => Ref.set(sent, input.messageId).pipe(Effect.as({} as never)), + getThreadShell: () => Effect.succeed(threadShell(projectId)), + getProjectThreadRecords: () => + Effect.map( + Ref.get(sent), + (userMessageId) => + ({ + runs: [ + { id: "run:1", status: "completed", userMessageId: "message:earlier" }, + { ...run, userMessageId }, + ], + turnItems: [ + { + type: "error", + status: "failed", + runId: "run:2", + nodeId: "node:root", + failure: { message: "Weekly usage limit reached" }, + }, + ], + }) as never, + ), + waitForThread: (input) => + Deferred.await(ended).pipe( + Effect.as({ + threadId: input.threadId, + run: { ...run, id: input.runId, userMessageId: "" }, + timedOut: false, + } as never), + ), + }), + Layer.mock(SecretRequests.SecretRequests)({}), + Layer.mock(ManagedProjectFolders.ManagedProjectFolders)({ + namedProjectsRoot: "/projects", + isScratchProject: () => Effect.succeed(false), + }), + NodeCrypto.layer, + Scheduler.layer, + ), + ), + ); + yield* Effect.gen(function* () { + const service = yield* ScheduledTaskService.ScheduledTaskService; + const { task } = yield* service.upsert({ + id: ScheduledTaskId.make("task:quota"), + title: "Quota bookkeeping", + prompt: "Update the notes", + enabled: true, + schedule: { type: "interval", everyMs: 60_000 }, + projectId, + threadId: ThreadId.make("thread:quota"), + workspaceStrategy: { type: "root" }, + modelSelection: { instanceId: ProviderInstanceId.make("codex"), model: "gpt-5.4" }, + runtimeMode: "full-access", + interactionMode: "default", + }); + yield* service.runNow({ id: task.id }); + const failed = yield* service.subscribeList().pipe( + Stream.map(({ tasks }) => tasks.find((candidate) => candidate.id === task.id)), + Stream.filter((current) => current?.lastRunStatus === "failed"), + Stream.runHead, + Effect.forkChild, + ); + yield* Deferred.succeed(ended, undefined); + const recorded = Option.getOrUndefined(yield* Fiber.join(failed)); + assert.equal(recorded?.lastRunError, "Weekly usage limit reached"); + assert.equal(recorded?.runCount, 1); + }).pipe(Effect.provide(layer)); + }).pipe(Effect.provide(SqlitePersistence.layerMemory)), +); + it.effect("a bound run follows its thread when promotion moves it mid-send", () => Effect.gen(function* () { const scratch = ProjectId.make("project:scratch"); diff --git a/apps/server/src/scheduledTasks/ScheduledTaskService.ts b/apps/server/src/scheduledTasks/ScheduledTaskService.ts index 146ee587fafdc..bf6c750d44c5f 100644 --- a/apps/server/src/scheduledTasks/ScheduledTaskService.ts +++ b/apps/server/src/scheduledTasks/ScheduledTaskService.ts @@ -57,6 +57,7 @@ import * as SecretRequests from "../secrets/SecretRequests.ts"; import * as ManagedPolicy from "../policy/ManagedPolicy.ts"; import * as DueHint from "../scheduling/DueHint.ts"; import * as Scheduler from "../scheduling/Scheduler.ts"; +import { latestRootProviderFailure } from "@t3tools/shared/orchestrationV2ThreadError"; import { isMissedFixedTimeRun, isSameSchedule, nextScheduledRunAt } from "./Schedule.ts"; import { redactHeaders, @@ -306,6 +307,7 @@ function iso(value: DateTime.DateTime): string { return DateTime.formatIso(DateTime.toUtc(value)); } +const RUN_OUTCOME_WAIT_MS = 24 * 60 * 60 * 1_000; const localNow = DateTime.withCurrentZoneLocal(DateTime.nowInCurrentZone); function nextRunAt( @@ -459,6 +461,7 @@ export const layer = Layer.effect( const mac = yield* Mac.Mac; const threadLaunch = yield* ThreadLaunchService.ThreadLaunchService; const threadManagement = yield* ThreadManagementService.ThreadManagementService; + const serviceScope = yield* Effect.scope; const managedFolders = yield* ManagedProjectFolders.ManagedProjectFolders; const secretRequests = yield* SecretRequests.SecretRequests; const scheduler = yield* Scheduler.Scheduler; @@ -863,6 +866,37 @@ export const layer = Layer.effect( ), ); + // A dispatch is recorded once the message is accepted; the provider can still + // reject the turn after that, so follow the run and record its failure. + const recordRunFailure = Effect.fn("ScheduledTaskService.recordRunFailure")(function* ( + taskId: ScheduledTaskId, + startedAtIso: string, + threadId: ThreadId, + messageId: MessageId, + ) { + 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); + if (dispatched === undefined) return; + const { run } = yield* threadManagement.waitForThread({ + ...thread, + runId: dispatched.id, + timeoutMs: RUN_OUTCOME_WAIT_MS, + }); + if (run?.status !== "failed") return; + const { turnItems } = yield* threadManagement.getProjectThreadRecords(thread, ["turnItems"]); + const error = latestRootProviderFailure(run, turnItems)?.message ?? "The run failed."; + 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* ( task: ScheduledTask, trigger: "scheduled" | "manual" | "webhook", @@ -1109,6 +1143,17 @@ export const layer = Layer.effect( ), ); if (written) yield* notifyChanged; + if (written && threadId !== null) { + yield* recordRunFailure(active.id, startedAtIso, threadId, messageId).pipe( + Effect.catchCause((cause) => + Effect.logWarning("Could not record scheduled run outcome", { + taskId: active.id, + cause, + }), + ), + Effect.forkIn(serviceScope), + ); + } // The stored row can belong to a replacement, so report this dispatch separately. return { task: recorded, threadId, dispatchFailed: !runSucceeded }; }).pipe( @@ -1922,8 +1967,6 @@ export const layer = Layer.effect( }); }); - const serviceScope = yield* Effect.scope; - const releaseFailedDelivery = Effect.fn("ScheduledTaskService.releaseFailedDelivery")( function* (deliveryId: string) { yield* releaseReplay(deliveryId); From 41e9ed60fad360df5c43fc73df6b6625309bba29 Mon Sep 17 00:00:00 2001 From: Julius Marminge Date: Thu, 12 Nov 2026 07:29:31 +0000 Subject: [PATCH 4/5] fix(server): a missed fixed-time slot cannot overwrite a newer run The due-task batch is a snapshot, so a run made while the scheduler waited on an earlier task could be stamped failed by that task's stale missed slot. Skip the reschedule when the row's next or last run time has moved on. Refs pingdotgg/t3code#30844 AI-assisted; verified with the tests and tools listed above. --- .../ScheduledTaskService.schedule.test.ts | 91 +++++++++++++++++++ .../scheduledTasks/ScheduledTaskService.ts | 8 +- 2 files changed, 97 insertions(+), 2 deletions(-) diff --git a/apps/server/src/scheduledTasks/ScheduledTaskService.schedule.test.ts b/apps/server/src/scheduledTasks/ScheduledTaskService.schedule.test.ts index 15a48147f28d4..92167ef83c5ca 100644 --- a/apps/server/src/scheduledTasks/ScheduledTaskService.schedule.test.ts +++ b/apps/server/src/scheduledTasks/ScheduledTaskService.schedule.test.ts @@ -465,3 +465,94 @@ it.effect("a fixed-time slot missed while the server was off shows on the task", }); }).pipe(Effect.provide(SqlitePersistence.layerMemory)), ); + +it.effect( + "a run made while the scheduler was busy is not overwritten by its stale missed slot", + () => + Effect.gen(function* () { + const at = (hour: number, minute = 0, day = 1) => + DateTime.makeZonedUnsafe( + { year: 2026, month: 7, day, hour, minute, second: 0, millisecond: 0 }, + { timeZone: DateTime.zoneMakeLocal(), adjustForTimeZone: true }, + ); + yield* TestClock.setTime(DateTime.toEpochMillis(at(6))); + const slowStarted = yield* Deferred.make(); + const releaseSlow = yield* Deferred.make(); + const lastProcessed = yield* Deferred.make(); + const layerDependencies = Layer.mergeAll( + NodeCrypto.layer, + Scheduler.layer, + ManagedPolicy.layerTest(), + Layer.mock(ThreadLaunchService.ThreadLaunchService)({ + launch: (input) => + (input.title === "Slow" + ? Deferred.succeed(slowStarted, undefined).pipe( + Effect.andThen(Deferred.await(releaseSlow)), + ) + : input.title === "Last" + ? Deferred.succeed(lastProcessed, undefined) + : Effect.void + ).pipe(Effect.as({ threadId: `thread-${input.title}`, resumed: false } as never)), + }), + Layer.mock(ManagedProjectFolders.ManagedProjectFolders)({ + namedProjectsRoot: "/projects", + isScratchProject: () => Effect.succeed(false), + }), + Layer.mock(ThreadManagementService.ThreadManagementService)({}), + Layer.mock(SecretRequests.SecretRequests)({}), + ); + const withServer = (effect: Effect.Effect) => + effect.pipe( + Effect.provide(ScheduledTaskService.layer.pipe(Layer.provide(layerDependencies))), + ); + const upsert = ( + title: string, + schedule: { type: "interval"; everyMs: number } | { type: "fixed_time"; timeOfDay: string }, + ) => + Effect.gen(function* () { + const service = yield* ScheduledTaskService.ScheduledTaskService; + const { task } = yield* service.upsert( + yield* decodeUpsertInput({ + title, + prompt: "Update the notes.", + enabled: true, + schedule, + projectId: "project-stale-slot", + workspaceStrategy: { type: "root" }, + modelSelection: { instanceId: "codex", model: "gpt-5.4" }, + runtimeMode: "full-access", + interactionMode: "default", + }), + ); + return task.id; + }); + + // Due at 07:00, 08:00 and 08:30; the server is off until 09:30. + const morningId = yield* withServer( + Effect.gen(function* () { + yield* upsert("Slow", { type: "interval", everyMs: 60 * 60 * 1000 }); + const id = yield* upsert("Morning", { type: "fixed_time", timeOfDay: "08:00" }); + yield* upsert("Last", { type: "interval", everyMs: 150 * 60 * 1000 }); + return id; + }), + ); + yield* TestClock.setTime(DateTime.toEpochMillis(at(9, 30))); + const morning = yield* withServer( + Effect.gen(function* () { + const service = yield* ScheduledTaskService.ScheduledTaskService; + yield* TestClock.adjust("6 seconds"); + yield* Deferred.await(slowStarted); + yield* service.runNow({ id: morningId }); + yield* Deferred.succeed(releaseSlow, undefined); + yield* Deferred.await(lastProcessed); + return (yield* service.list()).tasks.find((task) => task.id === morningId); + }), + ); + + expect({ status: morning?.lastRunStatus, error: morning?.lastRunError }).toEqual({ + status: "succeeded", + error: null, + }); + expect(morning?.nextRunAt).toBe(DateTime.formatIso(DateTime.toUtc(at(8, 0, 2)))); + }).pipe(Effect.provide(SqlitePersistence.layerMemory)), +); diff --git a/apps/server/src/scheduledTasks/ScheduledTaskService.ts b/apps/server/src/scheduledTasks/ScheduledTaskService.ts index bf6c750d44c5f..b62c212138995 100644 --- a/apps/server/src/scheduledTasks/ScheduledTaskService.ts +++ b/apps/server/src/scheduledTasks/ScheduledTaskService.ts @@ -1200,18 +1200,22 @@ export const layer = Layer.effect( missedRunAt: task.nextRunAt, rescheduledTo: next, }); - yield* sql` + // `task` can be a stale batch snapshot; a run or edit since then already moved the row on. + const updated = yield* sql` UPDATE scheduled_tasks SET next_run_at = ${next}, updated_at = ${iso(now)} ${runError === undefined ? sql`` : sql`, last_run_status = 'failed', last_run_error = ${runError}`} WHERE task_id = ${task.id} + AND next_run_at IS ${task.nextRunAt} + AND last_run_at IS ${task.lastRunAt} + RETURNING task_id `.pipe( Effect.mapError((cause) => taskError("Could not reschedule missed schedule task run.", { taskId: task.id, cause }), ), ); - yield* notifyChanged; + if (updated.length > 0) yield* notifyChanged; }); const runDueTasks = Effect.fn("ScheduledTaskService.runDueTasks")(function* () { From 7ffb54be496a10ce9dd962726acc8de760c824c0 Mon Sep 17 00:00:00 2001 From: Julius Marminge Date: Thu, 12 Nov 2026 07:41:39 +0000 Subject: [PATCH 5/5] fix(server): a stopped or cancelled scheduled run no longer reads as succeeded Record every unsuccessful ending of a dispatched run, not only a provider failure. A queued message promoted to a steer still counts as delivered. Refs pingdotgg/t3code#30844 AI-assisted; verified with the tests and tools listed above. --- .../ScheduledTaskService.test.ts | 11 +++++--- .../scheduledTasks/ScheduledTaskService.ts | 27 ++++++++++++++++--- 2 files changed, 32 insertions(+), 6 deletions(-) diff --git a/apps/server/src/scheduledTasks/ScheduledTaskService.test.ts b/apps/server/src/scheduledTasks/ScheduledTaskService.test.ts index 03cb9aec20407..ee4385ff2e0af 100644 --- a/apps/server/src/scheduledTasks/ScheduledTaskService.test.ts +++ b/apps/server/src/scheduledTasks/ScheduledTaskService.test.ts @@ -419,12 +419,16 @@ it.effect("thread cleanup notifies subscribers and a finishing run cannot recrea }).pipe(Effect.provide(SqlitePersistence.layerMemory)), ); -it.effect("a run the provider fails after accepting its message records the failure", () => +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) => Effect.gen(function* () { const projectId = ProjectId.make("project:test"); const ended = yield* Deferred.make(); const sent = yield* Ref.make(""); - const run = { id: "run:2", status: "failed", rootNodeId: "node:root" }; + const run = { id: "run:2", status: outcome.status, rootNodeId: "node:root" }; const layer = ScheduledTaskService.layer.pipe( Layer.provide( Layer.mergeAll( @@ -442,6 +446,7 @@ it.effect("a run the provider fails after accepting its message records the fail { id: "run:1", status: "completed", userMessageId: "message:earlier" }, { ...run, userMessageId }, ], + messages: [{ id: userMessageId, runId: "run:2" }], turnItems: [ { type: "error", @@ -496,7 +501,7 @@ it.effect("a run the provider fails after accepting its message records the fail ); yield* Deferred.succeed(ended, undefined); const recorded = Option.getOrUndefined(yield* Fiber.join(failed)); - assert.equal(recorded?.lastRunError, "Weekly usage limit reached"); + assert.equal(recorded?.lastRunError, outcome.error); assert.equal(recorded?.runCount, 1); }).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 b62c212138995..5af0aba39e6ed 100644 --- a/apps/server/src/scheduledTasks/ScheduledTaskService.ts +++ b/apps/server/src/scheduledTasks/ScheduledTaskService.ts @@ -308,6 +308,12 @@ function iso(value: DateTime.DateTime): string { } const RUN_OUTCOME_WAIT_MS = 24 * 60 * 60 * 1_000; +const UNSUCCESSFUL_RUN_ERRORS = { + failed: "The run failed.", + interrupted: "The run was stopped.", + cancelled: "The run was cancelled.", + rolled_back: "The run was rolled back.", +} as const; const localNow = DateTime.withCurrentZoneLocal(DateTime.nowInCurrentZone); function nextRunAt( @@ -885,9 +891,24 @@ export const layer = Layer.effect( runId: dispatched.id, timeoutMs: RUN_OUTCOME_WAIT_MS, }); - if (run?.status !== "failed") return; - const { turnItems } = yield* threadManagement.getProjectThreadRecords(thread, ["turnItems"]); - const error = latestRootProviderFailure(run, turnItems)?.message ?? "The run failed."; + 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 + ) + 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}