From 2fb8a65508ca013063c09f9fd22e8e89241fd8c6 Mon Sep 17 00:00:00 2001 From: Chris OHearn Date: Thu, 12 Nov 2026 13:59:57 -0700 Subject: [PATCH 1/6] fix(server): scheduled runs record their real outcome and refuse a stale project-folder workspace A scheduled run read as succeeded as soon as its launch was accepted, so a run that then failed workspace preparation showed succeeded until (and unless) the outcome follower flipped it, and a server restart lost that follower. Runs now record "started" on dispatch and "succeeded" or "failed" (with the error) once the run ends. Restarts resume following started runs. An unbound task MCP saved to run in its project folder because the folder was not a repository kept running there after `git init` or a clone (#69). The launch now refuses with a clear error, and an MCP update re-derives the workspace. Co-Authored-By: Claude Opus 5.5 --- .../src/mcp/OrchestratorMcpService.test.ts | 26 +- apps/server/src/mcp/OrchestratorMcpService.ts | 10 +- ...laudeAutomaticDelivery.integration.test.ts | 2 +- .../ThreadLaunchService.test.ts | 6 +- .../ScheduledTaskService.schedule.test.ts | 2 +- .../ScheduledTaskService.test.ts | 198 ++++++++++++- .../scheduledTasks/ScheduledTaskService.ts | 269 +++++++++++++----- .../ScheduledTaskService.webhook.test.ts | 8 +- .../chat/ThreadAutomationsPanel.tsx | 1 + .../settings/ScheduledTasksSettings.tsx | 2 +- packages/contracts/src/scheduledTask.ts | 9 +- 11 files changed, 428 insertions(+), 105 deletions(-) diff --git a/apps/server/src/mcp/OrchestratorMcpService.test.ts b/apps/server/src/mcp/OrchestratorMcpService.test.ts index f534194055336..e428fe9e0602b 100644 --- a/apps/server/src/mcp/OrchestratorMcpService.test.ts +++ b/apps/server/src/mcp/OrchestratorMcpService.test.ts @@ -3231,17 +3231,37 @@ describe("OrchestratorMcpService provider resolution", () => { }), ); - it.effect("keeps the saved workspace when an unbound task stays unbound", () => + it.effect.each([ + { + name: "keeps a project folder chosen in Settings", + task: { ...existing, creationSource: "web" }, + workspace: { type: "root" }, + }, + { + name: "keeps a saved worktree", + task: { + ...existing, + workspaceStrategy: { type: "worktree", baseRef: "develop", startFromOrigin: false }, + }, + workspace: { type: "worktree", baseRef: "develop", startFromOrigin: false }, + }, + { + // #69: saved as root while the folder was not a repository; it is one now. + name: "re-derives a project folder it saved, now that the folder is a repository", + task: existing, + workspace: { type: "worktree", baseRef: "origin/master", startFromOrigin: true }, + }, + ] as const)("an unbound task that stays unbound $name", ({ task, workspace }) => Effect.gen(function* () { const saved = yield* Ref.make>([]); const mcp = yield* OrchestratorMcpService.OrchestratorMcpService.pipe( - Effect.provide(service(masterRepository, saved)), + Effect.provide(service(masterRepository, saved, task)), ); yield* mcp.updateScheduledTask(client, { scheduledTaskId: existing.id, bindToCurrentThread: false, }); - assert.deepEqual(yield* Ref.get(saved), [{ type: "root" }]); + assert.deepEqual(yield* Ref.get(saved), [workspace]); }), ); diff --git a/apps/server/src/mcp/OrchestratorMcpService.ts b/apps/server/src/mcp/OrchestratorMcpService.ts index 66ede503ce363..0ad967d221efc 100644 --- a/apps/server/src/mcp/OrchestratorMcpService.ts +++ b/apps/server/src/mcp/OrchestratorMcpService.ts @@ -2169,9 +2169,15 @@ const make = Effect.gen(function* () { : input.bindToCurrentThread && parent !== undefined ? parent.thread.id : null; - // Reset the workspace only when the binding flips, so unbinding cannot leave runs in the shared checkout. + // Reset the workspace when the binding flips, so unbinding cannot leave runs in the shared + // checkout. An unbound root it derived is re-derived too: the folder may be a repository now. const workspaceStrategy = - (threadId === null) === (existing.threadId === null) + (threadId === null) === (existing.threadId === null) && + !( + threadId === null && + existing.creationSource === "mcp" && + existing.workspaceStrategy.type === "root" + ) ? existing.workspaceStrategy : yield* scheduledTaskWorkspaceStrategy(threadId !== null, existing.projectId); const upsertInput: ScheduledTaskUpsertInput = { diff --git a/apps/server/src/orchestration-v2/ClaudeAutomaticDelivery.integration.test.ts b/apps/server/src/orchestration-v2/ClaudeAutomaticDelivery.integration.test.ts index faa8d8b8b3cbf..6240ae6c0f448 100644 --- a/apps/server/src/orchestration-v2/ClaudeAutomaticDelivery.integration.test.ts +++ b/apps/server/src/orchestration-v2/ClaudeAutomaticDelivery.integration.test.ts @@ -326,7 +326,7 @@ it.effect.each(deliveries)( }), ); const ran = yield* service.runNow({ id: task.id }); - assert.equal(ran.task.lastRunStatus, "succeeded"); + assert.equal(ran.task.lastRunStatus, "started"); }).pipe( Effect.provide( ScheduledTaskService.layer.pipe( diff --git a/apps/server/src/orchestration-v2/ThreadLaunchService.test.ts b/apps/server/src/orchestration-v2/ThreadLaunchService.test.ts index bf37cdb29d1ea..363fcf61e6307 100644 --- a/apps/server/src/orchestration-v2/ThreadLaunchService.test.ts +++ b/apps/server/src/orchestration-v2/ThreadLaunchService.test.ts @@ -377,7 +377,7 @@ it.effect.each( creationSource: createdBy === "agent" ? "mcp" : "web", }); const result = yield* tasks.runNow({ id: task.id }); - assert.equal(result.task.lastRunStatus, "succeeded"); + assert.equal(result.task.lastRunStatus, "started"); const projectThreads = yield* threads.listProjectThreads({ projectId, includeSubagents: false, @@ -494,7 +494,7 @@ it.effect("runs a thread-bound automation on the provider the thread uses now", }); const result = yield* tasks.runNow({ id: task.id }); - assert.equal(result.task.lastRunStatus, "succeeded"); + assert.equal(result.task.lastRunStatus, "started"); const projection = yield* threads.getThreadProjection(launched.threadId); const message = projection.messages.find((item) => item.scheduledTaskId === task.id); const run = projection.runs.find((item) => item.id === message?.runId); @@ -1086,7 +1086,7 @@ it.effect("runs a Scratch scheduled task saved with a worktree in its own folder WHERE task_id = ${task.id} `; const ran = yield* tasks.runNow({ id: task.id }); - assert.equal(ran.task.lastRunStatus, "succeeded"); + assert.equal(ran.task.lastRunStatus, "started"); const [thread] = yield* threads.listProjectThreads({ projectId, includeSubagents: false }); assert.isDefined(thread); const projection = yield* preparationSettled(thread!.id); diff --git a/apps/server/src/scheduledTasks/ScheduledTaskService.schedule.test.ts b/apps/server/src/scheduledTasks/ScheduledTaskService.schedule.test.ts index 92167ef83c5ca..f27f820647734 100644 --- a/apps/server/src/scheduledTasks/ScheduledTaskService.schedule.test.ts +++ b/apps/server/src/scheduledTasks/ScheduledTaskService.schedule.test.ts @@ -550,7 +550,7 @@ it.effect( ); expect({ status: morning?.lastRunStatus, error: morning?.lastRunError }).toEqual({ - status: "succeeded", + status: "started", error: null, }); expect(morning?.nextRunAt).toBe(DateTime.formatIso(DateTime.toUtc(at(8, 0, 2)))); diff --git a/apps/server/src/scheduledTasks/ScheduledTaskService.test.ts b/apps/server/src/scheduledTasks/ScheduledTaskService.test.ts index 53ea639975bd2..b4edad2dc8958 100644 --- a/apps/server/src/scheduledTasks/ScheduledTaskService.test.ts +++ b/apps/server/src/scheduledTasks/ScheduledTaskService.test.ts @@ -2,6 +2,7 @@ import * as Scheduler from "../scheduling/Scheduler.ts"; import * as NodeUtil from "node:util"; import * as NodeCrypto from "@effect/platform-node/NodeCrypto"; +import * as NodeServices from "@effect/platform-node/NodeServices"; import { assert, it, vi } from "@effect/vitest"; import { ProjectId, @@ -15,9 +16,11 @@ import { import * as DateTime from "effect/DateTime"; import * as Effect from "effect/Effect"; import * as Exit from "effect/Exit"; +import * as FileSystem from "effect/FileSystem"; import * as Layer from "effect/Layer"; import * as Logger from "effect/Logger"; import * as Option from "effect/Option"; +import * as Path from "effect/Path"; import * as SqlClient from "effect/sql/SqlClient"; import * as Fiber from "effect/Fiber"; @@ -419,21 +422,40 @@ it.effect("thread cleanup notifies subscribers and a finishing run cannot recrea }).pipe(Effect.provide(SqlitePersistence.layerMemory)), ); +const PREPARATION_FAILURE = + "Workspace preparation failed during provision worktree: Git command failed in GitWorkflowService.resolveWorktreeBaseRemote (/home/user/app): Failed to resolve the VCS driver for this Git command."; + 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) => + { name: "ends failed", run: "failed", status: "failed", error: "Weekly usage limit reached" }, + { name: "ends interrupted", run: "interrupted", status: "failed", error: "The run was stopped." }, + { name: "ends cancelled", run: "cancelled", status: "failed", error: "The run was cancelled." }, + { name: "completes", run: "completed", status: "succeeded", error: null }, + // The Jarvis inbox heartbeat failed this way on its first launch and still read as succeeded. + { + name: "fails workspace preparation on its first launch", + run: "failed", + status: "failed", + error: PREPARATION_FAILURE, + unbound: true, + }, +] as const)("a run that $name after its message was accepted records its outcome", (outcome) => Effect.gen(function* () { const projectId = ProjectId.make("project:test"); + const threadId = ThreadId.make("thread:quota"); + const unbound = "unbound" in outcome; const ended = yield* Deferred.make(); const sent = yield* Ref.make(""); - const run = { id: "run:2", status: outcome.status, rootNodeId: "node:root" }; + const run = { id: "run:2", status: outcome.run, rootNodeId: "node:root" }; const layer = ScheduledTaskService.layer.pipe( Layer.provide( Layer.mergeAll( ManagedPolicy.layerTest(), - Layer.mock(ThreadLaunchService.ThreadLaunchService)({}), + Layer.mock(ThreadLaunchService.ThreadLaunchService)({ + launch: (input) => + Ref.set(sent, input.initialMessage?.messageId ?? "").pipe( + Effect.as({ threadId } as never), + ), + }), Layer.mock(ThreadManagementService.ThreadManagementService)({ sendToThread: (input) => Ref.set(sent, input.messageId).pipe(Effect.as({} as never)), getThreadShell: () => Effect.succeed(threadShell(projectId)), @@ -453,7 +475,9 @@ it.effect.each([ status: "failed", runId: "run:2", nodeId: "node:root", - failure: { message: "Weekly usage limit reached" }, + failure: { + message: unbound ? PREPARATION_FAILURE : "Weekly usage limit reached", + }, }, ], }) as never, @@ -486,27 +510,177 @@ it.effect.each([ enabled: true, schedule: { type: "interval", everyMs: 60_000 }, projectId, - threadId: ThreadId.make("thread:quota"), + threadId: unbound ? null : threadId, 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( + const dispatched = yield* service.runNow({ id: task.id }); + assert.equal(dispatched.task.lastRunStatus, "started"); + const settled = yield* service.subscribeList().pipe( Stream.map(({ tasks }) => tasks.find((candidate) => candidate.id === task.id)), - Stream.filter((current) => current?.lastRunStatus === "failed"), + Stream.filter((current) => current?.lastRunStatus === outcome.status), Stream.runHead, Effect.forkChild, ); yield* Deferred.succeed(ended, undefined); - const recorded = Option.getOrUndefined(yield* Fiber.join(failed)); + const recorded = Option.getOrUndefined(yield* Fiber.join(settled)); assert.equal(recorded?.lastRunError, outcome.error); assert.equal(recorded?.runCount, 1); }).pipe(Effect.provide(layer)); }).pipe(Effect.provide(SqlitePersistence.layerMemory)), ); +it.effect("a restart picks up each started run and records how it ended", () => + Effect.gen(function* () { + const sql = yield* SqlClient.SqlClient; + const startedAt = "2026-11-12T17:44:47.869Z"; + const threadId = ThreadId.make("thread:resumed"); + const messageId = `scheduled-task-message:task:resumed:${Date.parse(startedAt)}:scheduled`; + for (const id of ["task:resumed", "task:lost"]) { + yield* insertRow(sql, { id, next: null, enabled: 1, status: "started" }, startedAt); + yield* sql`UPDATE scheduled_tasks SET last_run_at = ${startedAt} WHERE task_id = ${id}`; + } + yield* sql` + INSERT INTO orchestration_v2_projection_messages + (message_id, thread_id, run_id, role, streaming, created_at, updated_at, payload_json) + VALUES (${messageId}, ${threadId}, 'run:1', 'user', 0, ${startedAt}, ${startedAt}, '{}') + `; + const run = { + id: "run:1", + status: "failed", + rootNodeId: "node:root", + userMessageId: messageId, + }; + const layer = ScheduledTaskService.layer.pipe( + Layer.provide( + Layer.mergeAll( + ManagedPolicy.layerTest(), + Layer.mock(ThreadLaunchService.ThreadLaunchService)({}), + Layer.mock(ThreadManagementService.ThreadManagementService)({ + getThreadShell: () => Effect.succeed(threadShell(ProjectId.make("project:test"))), + getProjectThreadRecords: () => + Effect.succeed({ + runs: [run], + messages: [{ id: messageId, runId: run.id }], + turnItems: [ + { + type: "error", + status: "failed", + runId: run.id, + nodeId: run.rootNodeId, + failure: { message: PREPARATION_FAILURE }, + }, + ], + } as never), + waitForThread: (input) => + Effect.succeed({ threadId: input.threadId, run, 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 settled = yield* service.subscribeList().pipe( + Stream.map(({ tasks }) => tasks), + Stream.filter((tasks) => tasks.every((task) => task.lastRunStatus === "failed")), + Stream.runHead, + ); + const errors = Object.fromEntries( + Option.getOrThrow(settled).map((task) => [task.id, task.lastRunError]), + ); + assert.deepEqual(errors, { + "task:resumed": PREPARATION_FAILURE, + "task:lost": "The run's thread was deleted before the run finished.", + }); + }).pipe(Effect.provide(layer)); + }).pipe(Effect.provide(SqlitePersistence.layerMemory)), +); + +// #69: a task MCP saved to run in its folder because the folder was not a repository. +it.effect.each([ + { name: "refuses an unbound MCP task", git: true, dispatch: "refused" }, + { name: "runs a bound task", git: true, bound: true, dispatch: "sent" }, + { name: "runs a task whose folder was chosen in Settings", git: true, source: "web" }, + { name: "runs a task in a scratch project", git: true, scratch: true }, + { name: "runs a task whose folder is still not a repository", git: false }, +] as const)("a saved project-folder workspace in a folder that became a repository $name", (row) => + Effect.gen(function* () { + const fileSystem = yield* FileSystem.FileSystem; + const path = yield* Path.Path; + const sql = yield* SqlClient.SqlClient; + const root = yield* fileSystem.makeTempDirectoryScoped({ prefix: "scheduled-root-" }); + if (row.git) yield* fileSystem.makeDirectory(path.join(root, ".git")); + const projectId = ProjectId.make("project:repo"); + yield* sql` + INSERT INTO projection_projects (project_id, title, workspace_root, scripts_json, created_at, updated_at) + VALUES (${projectId}, 'Repo', ${root}, '[]', '', '') + `; + const dispatched = yield* Ref.make<"launched" | "sent" | "refused">("refused"); + const layer = ScheduledTaskService.layer.pipe( + Layer.provide( + Layer.mergeAll( + ManagedPolicy.layerTest(), + Layer.mock(ThreadLaunchService.ThreadLaunchService)({ + launch: () => + Ref.set(dispatched, "launched").pipe( + Effect.as({ threadId: ThreadId.make("thread:launched") } as never), + ), + }), + Layer.mock(ThreadManagementService.ThreadManagementService)({ + sendToThread: () => Ref.set(dispatched, "sent").pipe(Effect.as({} as never)), + getThreadShell: () => Effect.succeed(threadShell(projectId)), + getProjectThreadRecords: () => + Effect.succeed({ runs: [], messages: [], turnItems: [] } as never), + }), + Layer.mock(SecretRequests.SecretRequests)({}), + Layer.mock(ManagedProjectFolders.ManagedProjectFolders)({ + namedProjectsRoot: "/projects", + isScratchProject: () => Effect.succeed("scratch" in row), + }), + NodeCrypto.layer, + Scheduler.layer, + ), + ), + ); + yield* Effect.gen(function* () { + const service = yield* ScheduledTaskService.ScheduledTaskService; + const { task } = yield* service.upsert({ + id: ScheduledTaskId.make("task:repo"), + title: "Nightly notes", + prompt: "Update the notes", + enabled: true, + schedule: { type: "interval", everyMs: 60_000 }, + projectId, + threadId: "bound" in row ? ThreadId.make("thread:bound") : null, + workspaceStrategy: { type: "root" }, + modelSelection: { instanceId: ProviderInstanceId.make("codex"), model: "gpt-5.4" }, + runtimeMode: "full-access", + interactionMode: "default", + createdBy: "agent", + creationSource: "source" in row ? row.source : "mcp", + }); + const result = yield* service.runNow({ id: task.id }); + assert.equal(yield* Ref.get(dispatched), "dispatch" in row ? row.dispatch : "launched"); + if ("dispatch" in row && row.dispatch === "refused") { + assert.equal(result.task.lastRunStatus, "failed"); + assert.include(result.task.lastRunError, `${root} is now a Git repository`); + assert.equal(result.task.runCount, 1); + } else { + assert.equal(result.task.lastRunStatus, "started"); + } + }).pipe(Effect.provide(layer)); + }).pipe(Effect.provide(Layer.merge(SqlitePersistence.layerMemory, NodeServices.layer))), +); + 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 74d519615bcc2..96fc99dcc7cf5 100644 --- a/apps/server/src/scheduledTasks/ScheduledTaskService.ts +++ b/apps/server/src/scheduledTasks/ScheduledTaskService.ts @@ -59,6 +59,7 @@ import * as DueHint from "../scheduling/DueHint.ts"; import * as ScheduledTaskChanges from "./ScheduledTaskChanges.ts"; import * as Scheduler from "../scheduling/Scheduler.ts"; import { latestRootProviderFailure } from "@t3tools/shared/orchestrationV2ThreadError"; +import { hasGitEntry } from "../mcp/gitEntryProbe.ts"; import { isMissedFixedTimeRun, isSameSchedule, nextScheduledRunAt } from "./Schedule.ts"; import { redactHeaders, @@ -234,6 +235,12 @@ interface ScheduledTaskRow { readonly webhook_secret: string | null; } +interface StartedRunRow { + readonly task_id: string; + readonly last_run_at: string; + readonly last_run_webhook_delivery_id: string | null; +} + interface WebhookDeliveryRow { readonly delivery_id: string; readonly task_id: string; @@ -777,7 +784,7 @@ export const layer = Layer.effect( readonly id: ScheduledTaskId; readonly completedAtIso: string; readonly nextRunAtIso: string | null; - readonly status: "succeeded" | "failed"; + readonly status: "started" | "failed"; readonly error: string | null; readonly startedAtIso: string; }) => @@ -880,52 +887,140 @@ 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* ( + const recordRunOutcome = ( 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 === 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` + status: "succeeded" | "failed", + error: string | null, + ) => + 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' + SET last_run_status = ${status}, last_run_error = ${error} + WHERE task_id = ${taskId} AND last_run_at = ${startedAtIso} AND last_run_status = 'started' RETURNING task_id - `; - if (updated.length > 0) yield* notifyChanged; + `.pipe(Effect.flatMap((rows) => (rows.length > 0 ? notifyChanged : Effect.void))); + + // A dispatch is recorded as started once the message is accepted; workspace preparation + // or the provider can still fail the run after that, so follow it to its outcome. + const followRunOutcome = Effect.fn("ScheduledTaskService.followRunOutcome")( + function* ( + taskId: ScheduledTaskId, + startedAtIso: string, + threadId: ThreadId, + messageId: MessageId, + ) { + const shell = yield* threadManagement.getThreadShell(threadId); + if (shell === null) { + return yield* recordRunOutcome( + taskId, + startedAtIso, + "failed", + "The run's thread was deleted before the run finished.", + ); + } + const thread = { projectId: shell.projectId, threadId }; + const { runs, messages } = yield* threadManagement.getProjectThreadRecords(thread, [ + "runs", + "messages", + ]); + let runId = + runs.find((run) => run.userMessageId === messageId)?.id ?? + messages.find((message) => message.id === messageId)?.runId ?? + null; + // A queued message promoted to a steer cancels its run and joins the active one. + for (let hops = 0; runId !== null && hops < 3; hops++) { + const { run } = yield* threadManagement.waitForThread({ + ...thread, + runId, + timeoutMs: RUN_OUTCOME_WAIT_MS, + }); + // Still active after the wait: the run has not finished, so it stays started. + if (run === null || !ThreadManagementService.isTerminalRunStatus(run.status)) return; + if (run.status === "completed") { + return yield* recordRunOutcome(taskId, startedAtIso, "succeeded", null); + } + const { turnItems, messages: current } = yield* threadManagement.getProjectThreadRecords( + thread, + ["turnItems", "messages"], + ); + const joined = current.find((message) => message.id === messageId)?.runId ?? null; + if (run.status === "cancelled" && joined !== null && joined !== run.id) { + runId = joined; + continue; + } + const error = + latestRootProviderFailure(run, turnItems)?.message ?? + UNSUCCESSFUL_RUN_ERRORS[run.status]; + return yield* recordRunOutcome(taskId, startedAtIso, "failed", error); + } + if (runId === null) { + yield* recordRunOutcome(taskId, startedAtIso, "failed", "The run never started."); + } + }, + (effect, taskId) => + effect.pipe( + Effect.catchCause((cause) => + Effect.logWarning("Could not record scheduled run outcome", { taskId, cause }), + ), + ), + ); + + // The message id is derived from the run's start time and trigger (see runTask's fireKey), + // so a restart can find a started run again and keep following it. + const resumeStartedRun = Effect.fn("ScheduledTaskService.resumeStartedRun")(function* ( + row: StartedRunRow, + ) { + const taskId = ScheduledTaskId.make(row.task_id); + const startedMs = Date.parse(row.last_run_at); + const fireKeys = + row.last_run_webhook_delivery_id !== null + ? [`${row.task_id}:webhook:${row.last_run_webhook_delivery_id}`] + : [`${row.task_id}:${startedMs}:scheduled`, `${row.task_id}:${startedMs}:manual`]; + const messageIds = fireKeys.map((key) => `scheduled-task-message:${key}`); + const [found] = yield* sql<{ message_id: string; thread_id: string }>` + SELECT message_id, thread_id FROM orchestration_v2_projection_messages + WHERE ${sql.in("message_id", messageIds)} + LIMIT 1 + `.pipe(NodeSqliteClient.readOnly); + if (found === undefined) { + return yield* recordRunOutcome( + taskId, + row.last_run_at, + "failed", + "The run's thread was deleted before the run finished.", + ); + } + yield* followRunOutcome( + taskId, + row.last_run_at, + ThreadId.make(found.thread_id), + MessageId.make(found.message_id), + ).pipe(Effect.forkIn(serviceScope)); }); + // Saved by MCP while its folder was not a repository (see OrchestratorMcpService's + // scheduledTaskWorkspaceStrategy). Running it after `git init` or a clone would put the agent + // in the shared checkout, so refuse until the task is saved again and its workspace re-derived. + const refuseStaleRootWorkspace = Effect.fn("ScheduledTaskService.refuseStaleRootWorkspace")( + function* (task: ScheduledTask) { + if ( + task.threadId !== null || + task.workspaceStrategy.type !== "root" || + task.creationSource !== "mcp" || + (yield* managedFolders.isScratchProject(task.projectId)) + ) + return; + const [project] = yield* sql<{ workspace_root: string }>` + SELECT workspace_root FROM projection_projects WHERE project_id = ${task.projectId} + `.pipe(NodeSqliteClient.readOnly); + if (project === undefined || !(yield* hasGitEntry(project.workspace_root))) return; + return yield* taskError( + `${project.workspace_root} is now a Git repository, but this task was saved to run directly in that folder because it was not one. Update the task (any change through schedule_task re-checks the folder) or choose a workspace in Settings.`, + { taskId: task.id }, + ); + }, + ); + const runTask = Effect.fn("ScheduledTaskService.runTask")(function* ( task: ScheduledTask, trigger: "scheduled" | "manual" | "webhook", @@ -958,11 +1053,7 @@ export const layer = Layer.effect( NodeSqliteClient.readOnly, )).length > 0 ) - return { - task: { ...active, lastRunStatus: "succeeded" as const }, - threadId: null, - dispatchFailed: false, - }; + return { task: active, threadId: null, dispatchFailed: false }; const parsedNextRunAt = active.nextRunAt === null ? Option.none() : DateTime.make(active.nextRunAt); if ( @@ -1069,30 +1160,34 @@ export const layer = Layer.effect( const result = active.threadId === null ? yield* Effect.exit( - threadLaunch - .launch({ - commandId, - projectId: active.projectId, - title: active.title, - modelSelection: active.modelSelection, - runtimeMode: active.runtimeMode, - interactionMode: active.interactionMode, - workspaceStrategy: active.workspaceStrategy, - initialMessage: { - messageId, - scheduledTaskId: active.id, - text: prompt, - attachments: [], - }, - createdBy: active.createdBy, - creationSource: active.creationSource, - }) - .pipe( - Effect.map((launched) => launched.threadId), - Effect.tap((threadId) => - threadMode === "new_thread" ? Effect.void : bindThread(active, threadId), - ), + refuseStaleRootWorkspace(active).pipe( + Effect.andThen( + threadLaunch + .launch({ + commandId, + projectId: active.projectId, + title: active.title, + modelSelection: active.modelSelection, + runtimeMode: active.runtimeMode, + interactionMode: active.interactionMode, + workspaceStrategy: active.workspaceStrategy, + initialMessage: { + messageId, + scheduledTaskId: active.id, + text: prompt, + attachments: [], + }, + createdBy: active.createdBy, + creationSource: active.creationSource, + }) + .pipe( + Effect.map((launched) => launched.threadId), + Effect.tap((threadId) => + threadMode === "new_thread" ? Effect.void : bindThread(active, threadId), + ), + ), ), + ), ) : yield* Effect.exit( // The stored selection only seeds fresh threads. A bound run @@ -1105,7 +1200,7 @@ export const layer = Layer.effect( const runSucceeded = result._tag === "Success"; if (webhook !== undefined && runSucceeded) yield* Ref.set(webhook.dispatched, true); const completedAt = yield* localNow; - const lastRunStatus = runSucceeded ? ("succeeded" as const) : ("failed" as const); + const lastRunStatus = runSucceeded ? ("started" as const) : ("failed" as const); const lastRunError = runSucceeded ? null : errorMessage(result.cause); const threadId = runSucceeded ? result.value : null; // One transaction, so an edit cannot land between reading the schedule and @@ -1173,13 +1268,7 @@ 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, - }), - ), + yield* followRunOutcome(active.id, startedAtIso, threadId, messageId).pipe( Effect.forkIn(serviceScope), ); } @@ -1351,6 +1440,32 @@ export const layer = Layer.effect( ), ); + // A run's outcome follower does not survive a restart; pick each started run up again. + yield* sql` + SELECT task_id, last_run_at, last_run_webhook_delivery_id FROM scheduled_tasks + WHERE last_run_status = 'started' AND last_run_at IS NOT NULL + `.pipe( + NodeSqliteClient.readOnly, + Effect.flatMap((rows) => + Effect.forEach( + rows, + (row) => + resumeStartedRun(row).pipe( + Effect.catch((cause) => + Effect.logWarning("Could not resume following a started schedule task run", { + taskId: row.task_id, + cause, + }), + ), + ), + { discard: true }, + ), + ), + Effect.catch((cause) => + Effect.logWarning("Could not resume following started schedule task runs", { cause }), + ), + ); + yield* scheduler.register("scheduled-tasks", runDueTasks()); const list: ScheduledTaskService["Service"]["list"] = () => diff --git a/apps/server/src/scheduledTasks/ScheduledTaskService.webhook.test.ts b/apps/server/src/scheduledTasks/ScheduledTaskService.webhook.test.ts index e06ed817ac4f4..a216e3a91923e 100644 --- a/apps/server/src/scheduledTasks/ScheduledTaskService.webhook.test.ts +++ b/apps/server/src/scheduledTasks/ScheduledTaskService.webhook.test.ts @@ -1082,7 +1082,7 @@ it.effect("retrying a running window keeps the prompt it was dispatched with", ( yield* service.triggerWebhook(requestFor(task)); yield* save("Prompt at dispatch"); yield* sql`CREATE TRIGGER fail_completion BEFORE UPDATE ON scheduled_tasks - WHEN NEW.last_run_status = 'succeeded' BEGIN SELECT RAISE(FAIL, 'completion unavailable'); END`; + WHEN NEW.last_run_status = 'started' BEGIN SELECT RAISE(FAIL, 'completion unavailable'); END`; yield* TestClock.adjust("10 seconds"); yield* drainPendingDispatch(workers); const launched = yield* Queue.take(launches); @@ -1408,7 +1408,7 @@ it.effect("start failures release both replay keys and relay claims for redelive ); assert.equal(retry._tag === "accepted" && retry.outcome, "accepted"); yield* Queue.take(launches); - assert.equal((yield* service.list()).tasks[0]!.lastRunStatus, "succeeded"); + assert.equal((yield* service.list()).tasks[0]!.lastRunStatus, "started"); }), { failLaunches: { remaining: 1 } }, ), @@ -1528,7 +1528,7 @@ it.effect("keeps replay keys when dispatch succeeds but completion bookkeeping f }), ); yield* sql`CREATE TRIGGER fail_completion BEFORE UPDATE ON scheduled_tasks - WHEN NEW.last_run_status = 'succeeded' BEGIN SELECT RAISE(FAIL, 'completion unavailable'); END`; + WHEN NEW.last_run_status = 'started' BEGIN SELECT RAISE(FAIL, 'completion unavailable'); END`; yield* waitForSpan( service.triggerWebhook(requestFor(task)), "ScheduledTaskService.runWebhookDelivery", @@ -1920,7 +1920,7 @@ it.effect.each([false, true])( requestFor(task, { headers: { group: "old" }, bodyText: '{"value":"old"}' }), ); yield* sql`CREATE TRIGGER fail_old_completion BEFORE UPDATE ON scheduled_tasks - WHEN NEW.last_run_status = 'succeeded' BEGIN SELECT RAISE(FAIL, 'completion unavailable'); END`; + WHEN NEW.last_run_status = 'started' BEGIN SELECT RAISE(FAIL, 'completion unavailable'); END`; yield* TestClock.adjust("10 seconds"); yield* drainPendingDispatch(workers); yield* Queue.take(launches); diff --git a/apps/web/src/components/chat/ThreadAutomationsPanel.tsx b/apps/web/src/components/chat/ThreadAutomationsPanel.tsx index 1bec327103211..93e87a305c45c 100644 --- a/apps/web/src/components/chat/ThreadAutomationsPanel.tsx +++ b/apps/web/src/components/chat/ThreadAutomationsPanel.tsx @@ -34,6 +34,7 @@ import { const STATUS_DOT_CLASS: Record = { never: "bg-muted-foreground/40", running: "bg-sky-500", + started: "bg-sky-500", succeeded: "bg-emerald-500", failed: "bg-destructive", }; diff --git a/apps/web/src/components/settings/ScheduledTasksSettings.tsx b/apps/web/src/components/settings/ScheduledTasksSettings.tsx index 848b88130f48a..9eb1dc33a0af2 100644 --- a/apps/web/src/components/settings/ScheduledTasksSettings.tsx +++ b/apps/web/src/components/settings/ScheduledTasksSettings.tsx @@ -364,7 +364,7 @@ function deliveryOutcomeVariant(outcome: string) { function statusVariant(status: ScheduledTask["lastRunStatus"]) { if (status === "failed") return "error"; if (status === "succeeded") return "success"; - if (status === "running") return "info"; + if (status === "running" || status === "started") return "info"; return "outline"; } diff --git a/packages/contracts/src/scheduledTask.ts b/packages/contracts/src/scheduledTask.ts index 4297ad94b73e9..1c139aa0a6a8e 100644 --- a/packages/contracts/src/scheduledTask.ts +++ b/packages/contracts/src/scheduledTask.ts @@ -214,7 +214,14 @@ export const ScheduledTaskUpsertSchedule = Schema.Union([ ]); export type ScheduledTaskUpsertSchedule = typeof ScheduledTaskUpsertSchedule.Type; -export const ScheduledTaskRunStatus = Schema.Literals(["never", "running", "succeeded", "failed"]); +/** `running` while the run is being dispatched; `started` once it was accepted and its outcome is pending. */ +export const ScheduledTaskRunStatus = Schema.Literals([ + "never", + "running", + "started", + "succeeded", + "failed", +]); export type ScheduledTaskRunStatus = typeof ScheduledTaskRunStatus.Type; export const ScheduledTaskWebhookEndpoint = Schema.Struct({ From 3f78bf9f898dcebaf05b3cda9a36af242731f56d Mon Sep 17 00:00:00 2001 From: Chris OHearn Date: Thu, 12 Nov 2026 14:03:43 -0700 Subject: [PATCH 2/6] fix(server): follow a steered scheduled run past a restart continuation Co-Authored-By: Claude Opus 5.5 --- .../scheduledTasks/ScheduledTaskService.test.ts | 17 +++++++++++++++-- .../src/scheduledTasks/ScheduledTaskService.ts | 17 +++++++++++++---- 2 files changed, 28 insertions(+), 6 deletions(-) diff --git a/apps/server/src/scheduledTasks/ScheduledTaskService.test.ts b/apps/server/src/scheduledTasks/ScheduledTaskService.test.ts index b4edad2dc8958..ba55409dbecf8 100644 --- a/apps/server/src/scheduledTasks/ScheduledTaskService.test.ts +++ b/apps/server/src/scheduledTasks/ScheduledTaskService.test.ts @@ -430,6 +430,13 @@ it.effect.each([ { name: "ends interrupted", run: "interrupted", status: "failed", error: "The run was stopped." }, { name: "ends cancelled", run: "cancelled", status: "failed", error: "The run was cancelled." }, { name: "completes", run: "completed", status: "succeeded", error: null }, + { + name: "is steered into the active run, which completes", + run: "cancelled", + status: "succeeded", + error: null, + joins: true, + }, // The Jarvis inbox heartbeat failed this way on its first launch and still read as succeeded. { name: "fails workspace preparation on its first launch", @@ -443,6 +450,7 @@ it.effect.each([ const projectId = ProjectId.make("project:test"); const threadId = ThreadId.make("thread:quota"); const unbound = "unbound" in outcome; + const joins = "joins" in outcome; const ended = yield* Deferred.make(); const sent = yield* Ref.make(""); const run = { id: "run:2", status: outcome.run, rootNodeId: "node:root" }; @@ -468,7 +476,7 @@ it.effect.each([ { id: "run:1", status: "completed", userMessageId: "message:earlier" }, { ...run, userMessageId }, ], - messages: [{ id: userMessageId, runId: "run:2" }], + messages: [{ id: userMessageId, runId: joins ? "run:3" : "run:2" }], turnItems: [ { type: "error", @@ -486,7 +494,12 @@ it.effect.each([ Deferred.await(ended).pipe( Effect.as({ threadId: input.threadId, - run: { ...run, id: input.runId, userMessageId: "" }, + run: { + ...run, + id: input.runId, + status: joins && input.runId === "run:3" ? "completed" : run.status, + userMessageId: "", + }, timedOut: false, } as never), ), diff --git a/apps/server/src/scheduledTasks/ScheduledTaskService.ts b/apps/server/src/scheduledTasks/ScheduledTaskService.ts index 96fc99dcc7cf5..4c4c3a310c686 100644 --- a/apps/server/src/scheduledTasks/ScheduledTaskService.ts +++ b/apps/server/src/scheduledTasks/ScheduledTaskService.ts @@ -944,7 +944,13 @@ export const layer = Layer.effect( ["turnItems", "messages"], ); const joined = current.find((message) => message.id === messageId)?.runId ?? null; - if (run.status === "cancelled" && joined !== null && joined !== run.id) { + // waitForThread answers with a restart's continuation, so compare against both ids. + if ( + run.status === "cancelled" && + joined !== null && + joined !== runId && + joined !== run.id + ) { runId = joined; continue; } @@ -953,9 +959,12 @@ export const layer = Layer.effect( UNSUCCESSFUL_RUN_ERRORS[run.status]; return yield* recordRunOutcome(taskId, startedAtIso, "failed", error); } - if (runId === null) { - yield* recordRunOutcome(taskId, startedAtIso, "failed", "The run never started."); - } + yield* recordRunOutcome( + taskId, + startedAtIso, + "failed", + runId === null ? "The run never started." : "The run was cancelled.", + ); }, (effect, taskId) => effect.pipe( From c720472ab6dfca0930bc204525a7aa25a0e46e13 Mon Sep 17 00:00:00 2001 From: Chris OHearn Date: Thu, 12 Nov 2026 14:48:13 -0700 Subject: [PATCH 3/6] fix(server): track pending scheduled runs in memory instead of a new stored status The `started` literal broke clients and older builds that decode the four known statuses strictly. Stored and wire statuses are back to the four; an accepted run writes back the status from before it, and an optional `runInProgressSince` (never stored) marks a run still in progress. MCP still reports `started` through its own status schema. A restart finds pending runs again from their message ids. Also: a soft-deleted thread, a missing thread or a removed run records a failed outcome; a recorded missed slot is not overwritten by an older pending run; edits keep the stored creationSource so MCP-derived workspaces stay rechecked; the #69 refusal names the `.git` it found and warns when it is in a parent folder. Co-Authored-By: Claude Opus 5.5 --- apps/server/src/mcp/OrchestratorMcpService.ts | 3 +- apps/server/src/mcp/gitEntryProbe.ts | 29 +- .../src/mcp/toolkits/thread/handlers.ts | 3 +- apps/server/src/mcp/toolkits/thread/tools.ts | 3 +- ...laudeAutomaticDelivery.integration.test.ts | 2 +- .../ThreadLaunchService.test.ts | 6 +- .../ScheduledTaskService.schedule.test.ts | 117 ++++- .../ScheduledTaskService.test.ts | 444 ++++++++++++------ .../scheduledTasks/ScheduledTaskService.ts | 306 +++++++----- .../ScheduledTaskService.webhook.test.ts | 9 +- .../chat/ThreadAutomationsPanel.tsx | 4 +- .../settings/ScheduledTasksSettings.tsx | 8 +- packages/contracts/src/orchestratorMcp.ts | 16 +- packages/contracts/src/scheduledTask.ts | 22 +- 14 files changed, 688 insertions(+), 284 deletions(-) diff --git a/apps/server/src/mcp/OrchestratorMcpService.ts b/apps/server/src/mcp/OrchestratorMcpService.ts index 0ad967d221efc..ce10e8592bb14 100644 --- a/apps/server/src/mcp/OrchestratorMcpService.ts +++ b/apps/server/src/mcp/OrchestratorMcpService.ts @@ -36,6 +36,7 @@ import { type OrchestratorMcpDeleteScheduledTaskResult, type OrchestratorMcpListScheduledTasksResult, type OrchestratorMcpScheduledTask, + orchestratorMcpScheduledTaskRunStatus, type OrchestratorMcpScheduleTaskInput, type OrchestratorMcpScheduleTaskResult, type OrchestratorMcpTarget, @@ -293,7 +294,7 @@ function scheduledTaskSummary(task: ScheduledTask, mayRun: boolean): Orchestrato freshSession: scheduledTaskThreadMode(task) === "fresh_session", schedule: task.schedule, nextRunAt: task.nextRunAt, - lastRunStatus: task.lastRunStatus, + lastRunStatus: orchestratorMcpScheduledTaskRunStatus(task), ...(task.webhook?.url == null || !mayRun || task.blockedByPolicy !== undefined ? {} : { webhookUrl: task.webhook.url }), diff --git a/apps/server/src/mcp/gitEntryProbe.ts b/apps/server/src/mcp/gitEntryProbe.ts index e21b18975cb23..63875cbd94aeb 100644 --- a/apps/server/src/mcp/gitEntryProbe.ts +++ b/apps/server/src/mcp/gitEntryProbe.ts @@ -3,20 +3,33 @@ import * as NodeFS from "node:fs"; import * as NodePath from "node:path"; import * as Effect from "effect/Effect"; +export interface GitEntry { + readonly path: string; + /** The folder holding it, which is `cwd` itself or one of its parents. */ + readonly folder: string; + readonly inParent: boolean; +} + /** - * Whether `cwd` or any parent folder has a `.git` entry (folder or gitfile). A folder that cannot be - * inspected counts as having one, so callers that use this to prove "no repository" fail closed. + * The nearest `.git` entry (folder or gitfile) in `cwd` or any parent folder, or null. A folder that + * cannot be inspected counts as having one, so callers that use this to prove "no repository" fail + * closed. */ -export const hasGitEntry = (cwd: string): Effect.Effect => +export const findGitEntry = (cwd: string): Effect.Effect => Effect.sync(() => { - for (let folder = NodePath.resolve(cwd); ; folder = NodePath.dirname(folder)) { + const start = NodePath.resolve(cwd); + for (let folder = start; ; folder = NodePath.dirname(folder)) { + const found = { path: NodePath.join(folder, ".git"), folder, inParent: folder !== start }; try { - NodeFS.lstatSync(NodePath.join(folder, ".git")); - return true; + NodeFS.lstatSync(found.path); + return found; } catch (error) { const code = (error as NodeJS.ErrnoException).code; - if (code !== "ENOENT" && code !== "ENOTDIR") return true; + if (code !== "ENOENT" && code !== "ENOTDIR") return found; } - if (NodePath.dirname(folder) === folder) return false; + if (NodePath.dirname(folder) === folder) return null; } }); + +export const hasGitEntry = (cwd: string): Effect.Effect => + Effect.map(findGitEntry(cwd), (entry) => entry !== null); diff --git a/apps/server/src/mcp/toolkits/thread/handlers.ts b/apps/server/src/mcp/toolkits/thread/handlers.ts index efb56f962a205..860c8a01f04db 100644 --- a/apps/server/src/mcp/toolkits/thread/handlers.ts +++ b/apps/server/src/mcp/toolkits/thread/handlers.ts @@ -7,6 +7,7 @@ import { type OrchestrationV2ThreadProjection, type RunId, OrchestratorMcpFailure, + orchestratorMcpScheduledTaskRunStatus, type OrchestrationV2Command, } from "@t3tools/contracts"; import * as Effect from "effect/Effect"; @@ -130,7 +131,7 @@ export const layer = McpToolAccess.toLayer(ThreadToolkit, { : { link: McpInvocationContext.threadLinkFor(scope, threadId), }), - lastRunStatus: task.lastRunStatus, + lastRunStatus: orchestratorMcpScheduledTaskRunStatus(task), runCount: task.runCount, nextRunAt: task.nextRunAt, }; diff --git a/apps/server/src/mcp/toolkits/thread/tools.ts b/apps/server/src/mcp/toolkits/thread/tools.ts index 13d5f99960e39..426968e83dbf2 100644 --- a/apps/server/src/mcp/toolkits/thread/tools.ts +++ b/apps/server/src/mcp/toolkits/thread/tools.ts @@ -14,6 +14,7 @@ import { ProviderUserInputAnswers, IsoDateTime, OrchestratorMcpFailure, + OrchestratorMcpScheduledTaskRunStatus, OrchestrationV2DispatchCommandResult, ThreadId, RunId, @@ -268,7 +269,7 @@ const ScheduledTaskRunTool = Tool.make("run_scheduled_task_now", { taskId: ScheduledTaskId, threadId: ScheduledTask.fields.threadId, link: Schema.optional(Schema.String), - lastRunStatus: ScheduledTask.fields.lastRunStatus, + lastRunStatus: OrchestratorMcpScheduledTaskRunStatus, runCount: NonNegativeInt, nextRunAt: ScheduledTask.fields.nextRunAt, }), diff --git a/apps/server/src/orchestration-v2/ClaudeAutomaticDelivery.integration.test.ts b/apps/server/src/orchestration-v2/ClaudeAutomaticDelivery.integration.test.ts index 6240ae6c0f448..ff4cab2bfa8cd 100644 --- a/apps/server/src/orchestration-v2/ClaudeAutomaticDelivery.integration.test.ts +++ b/apps/server/src/orchestration-v2/ClaudeAutomaticDelivery.integration.test.ts @@ -326,7 +326,7 @@ it.effect.each(deliveries)( }), ); const ran = yield* service.runNow({ id: task.id }); - assert.equal(ran.task.lastRunStatus, "started"); + assert.equal(ran.task.runInProgressSince, ran.task.lastRunAt); }).pipe( Effect.provide( ScheduledTaskService.layer.pipe( diff --git a/apps/server/src/orchestration-v2/ThreadLaunchService.test.ts b/apps/server/src/orchestration-v2/ThreadLaunchService.test.ts index 363fcf61e6307..e60d6e6b7496d 100644 --- a/apps/server/src/orchestration-v2/ThreadLaunchService.test.ts +++ b/apps/server/src/orchestration-v2/ThreadLaunchService.test.ts @@ -377,7 +377,7 @@ it.effect.each( creationSource: createdBy === "agent" ? "mcp" : "web", }); const result = yield* tasks.runNow({ id: task.id }); - assert.equal(result.task.lastRunStatus, "started"); + assert.equal(result.task.runInProgressSince, result.task.lastRunAt); const projectThreads = yield* threads.listProjectThreads({ projectId, includeSubagents: false, @@ -494,7 +494,7 @@ it.effect("runs a thread-bound automation on the provider the thread uses now", }); const result = yield* tasks.runNow({ id: task.id }); - assert.equal(result.task.lastRunStatus, "started"); + assert.equal(result.task.runInProgressSince, result.task.lastRunAt); const projection = yield* threads.getThreadProjection(launched.threadId); const message = projection.messages.find((item) => item.scheduledTaskId === task.id); const run = projection.runs.find((item) => item.id === message?.runId); @@ -1086,7 +1086,7 @@ it.effect("runs a Scratch scheduled task saved with a worktree in its own folder WHERE task_id = ${task.id} `; const ran = yield* tasks.runNow({ id: task.id }); - assert.equal(ran.task.lastRunStatus, "started"); + assert.equal(ran.task.runInProgressSince, ran.task.lastRunAt); const [thread] = yield* threads.listProjectThreads({ projectId, includeSubagents: false }); assert.isDefined(thread); const projection = yield* preparationSettled(thread!.id); diff --git a/apps/server/src/scheduledTasks/ScheduledTaskService.schedule.test.ts b/apps/server/src/scheduledTasks/ScheduledTaskService.schedule.test.ts index f27f820647734..3e1bb4d001011 100644 --- a/apps/server/src/scheduledTasks/ScheduledTaskService.schedule.test.ts +++ b/apps/server/src/scheduledTasks/ScheduledTaskService.schedule.test.ts @@ -6,14 +6,18 @@ import { type OrchestrationV2Run, MessageId, RunId, + type ScheduledTask, ScheduledTaskUpsertInput, } from "@t3tools/contracts"; import * as DateTime from "effect/DateTime"; import * as Deferred from "effect/Deferred"; import * as Effect from "effect/Effect"; import * as Layer from "effect/Layer"; +import * as Queue from "effect/Queue"; import * as Ref from "effect/Ref"; import * as Schema from "effect/Schema"; +import * as Stream from "effect/Stream"; +import * as SqlClient from "effect/sql/SqlClient"; import * as TestClock from "effect/testing/TestClock"; import * as ThreadLaunchService from "../orchestration-v2/ThreadLaunchService.ts"; @@ -466,6 +470,116 @@ it.effect("a fixed-time slot missed while the server was off shows on the task", }).pipe(Effect.provide(SqlitePersistence.layerMemory)), ); +it.effect( + "a slot missed during a restart is not overwritten by the run still going from before", + () => + Effect.gen(function* () { + const at = (hour: number, minute = 0) => + DateTime.makeZonedUnsafe( + { year: 2026, month: 7, day: 1, hour, minute, second: 0, millisecond: 0 }, + { timeZone: DateTime.zoneMakeLocal(), adjustForTimeZone: true }, + ); + yield* TestClock.setTime(DateTime.toEpochMillis(at(8))); + const sql = yield* SqlClient.SqlClient; + const sent = yield* Ref.make(""); + const end = yield* Deferred.make(); + const threadId = "thread-pending-run"; + const layer = ScheduledTaskService.layer.pipe( + Layer.provide( + Layer.mergeAll( + NodeCrypto.layer, + Scheduler.layer, + ManagedPolicy.layerTest(), + Layer.mock(ThreadLaunchService.ThreadLaunchService)({ + launch: (input) => + Ref.set(sent, input.initialMessage?.messageId ?? "").pipe( + Effect.as({ threadId } as never), + ), + }), + Layer.mock(ManagedProjectFolders.ManagedProjectFolders)({ + namedProjectsRoot: "/projects", + isScratchProject: () => Effect.succeed(false), + }), + Layer.mock(ThreadManagementService.ThreadManagementService)({ + getThreadShell: () => + Effect.succeed({ projectId: "project-pending-run", deletedAt: null } as never), + getProjectThreadRecords: () => + Effect.map( + Ref.get(sent), + (userMessageId) => + ({ + runs: [{ id: "run:1", status: "running", userMessageId }], + messages: [{ id: userMessageId, runId: "run:1" }], + }) as never, + ), + waitForThread: (input) => + Deferred.await(end).pipe( + Effect.as({ + threadId: input.threadId, + run: { id: "run:1", status: "completed", rootNodeId: "node:root" }, + timedOut: false, + } as never), + ), + }), + Layer.mock(SecretRequests.SecretRequests)({}), + ), + ), + ); + + const taskId = yield* 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-pending-run", + workspaceStrategy: { type: "root" }, + modelSelection: { instanceId: "codex", model: "gpt-5.4" }, + runtimeMode: "full-access", + interactionMode: "default", + }), + ); + yield* service.runNow({ id: task.id }); + return task.id; + }).pipe(Effect.provide(layer)); + // What the launch committed; the mocked launch writes no projection. + yield* sql` + INSERT INTO orchestration_v2_projection_messages + (message_id, thread_id, run_id, role, streaming, created_at, updated_at, payload_json) + VALUES (${yield* Ref.get(sent)}, ${threadId}, 'run:1', 'user', 0, '', '', '{}') + `; + + // The server is off across the 09:00 slot while the 08:00 run keeps going. + yield* TestClock.setTime(DateTime.toEpochMillis(at(9, 30))); + yield* Effect.gen(function* () { + const service = yield* ScheduledTaskService.ScheduledTaskService; + const states = yield* Queue.unbounded(); + yield* service.subscribeList().pipe( + Stream.runForEach(({ tasks }) => + Queue.offer( + states, + tasks.find((candidate) => candidate.id === taskId), + ), + ), + Effect.forkChild, + ); + expect((yield* Queue.take(states))?.runInProgressSince).toBeDefined(); + yield* TestClock.adjust("6 seconds"); + const missed = yield* Queue.take(states); + expect(missed?.lastRunStatus).toBe("failed"); + yield* Deferred.succeed(end, undefined); + const afterRun = yield* Queue.take(states); + expect({ + status: afterRun?.lastRunStatus, + error: afterRun?.lastRunError, + since: afterRun?.runInProgressSince, + }).toEqual({ status: "failed", error: missed?.lastRunError, since: undefined }); + }).pipe(Effect.provide(layer)); + }).pipe(Effect.provide(SqlitePersistence.layerMemory)), +); + it.effect( "a run made while the scheduler was busy is not overwritten by its stale missed slot", () => @@ -549,8 +663,9 @@ it.effect( }), ); + // The manual run's outcome is still pending, and the stale slot recorded no miss over it. expect({ status: morning?.lastRunStatus, error: morning?.lastRunError }).toEqual({ - status: "started", + status: "never", error: null, }); expect(morning?.nextRunAt).toBe(DateTime.formatIso(DateTime.toUtc(at(8, 0, 2)))); diff --git a/apps/server/src/scheduledTasks/ScheduledTaskService.test.ts b/apps/server/src/scheduledTasks/ScheduledTaskService.test.ts index ba55409dbecf8..1170e8850a89f 100644 --- a/apps/server/src/scheduledTasks/ScheduledTaskService.test.ts +++ b/apps/server/src/scheduledTasks/ScheduledTaskService.test.ts @@ -7,6 +7,7 @@ import { assert, it, vi } from "@effect/vitest"; import { ProjectId, ProviderInstanceId, + type ScheduledTask, ScheduledTaskError, ScheduledTaskId, type ScheduledTaskUpsertInput, @@ -26,6 +27,7 @@ import * as SqlClient from "effect/sql/SqlClient"; import * as Fiber from "effect/Fiber"; import * as Stream from "effect/Stream"; import * as Deferred from "effect/Deferred"; +import * as Queue from "effect/Queue"; import * as Ref from "effect/Ref"; import * as Schema from "effect/Schema"; import * as TestClock from "effect/testing/TestClock"; @@ -424,12 +426,44 @@ it.effect("thread cleanup notifies subscribers and a finishing run cannot recrea const PREPARATION_FAILURE = "Workspace preparation failed during provision worktree: Git command failed in GitWorkflowService.resolveWorktreeBaseRemote (/home/user/app): Failed to resolve the VCS driver for this Git command."; +const THREAD_DELETED = "The run's thread was deleted before the run finished."; + +const followerLayer = ( + threadManagement: Layer.Layer, + threadLaunch: Layer.Layer = Layer.mock( + ThreadLaunchService.ThreadLaunchService, + )({}), +) => + ScheduledTaskService.layer.pipe( + Layer.provide( + Layer.mergeAll( + ManagedPolicy.layerTest(), + threadLaunch, + threadManagement, + Layer.mock(SecretRequests.SecretRequests)({}), + Layer.mock(ManagedProjectFolders.ManagedProjectFolders)({ + namedProjectsRoot: "/projects", + isScratchProject: () => Effect.succeed(false), + }), + NodeCrypto.layer, + Scheduler.layer, + ), + ), + ); it.effect.each([ { name: "ends failed", run: "failed", status: "failed", error: "Weekly usage limit reached" }, { name: "ends interrupted", run: "interrupted", status: "failed", error: "The run was stopped." }, { name: "ends cancelled", run: "cancelled", status: "failed", error: "The run was cancelled." }, { name: "completes", run: "completed", status: "succeeded", error: null }, + // The Jarvis inbox heartbeat failed this way on its first launch and still read as succeeded. + { + name: "fails workspace preparation on its first launch", + run: "failed", + status: "failed", + error: PREPARATION_FAILURE, + unbound: true, + }, { name: "is steered into the active run, which completes", run: "cancelled", @@ -437,13 +471,26 @@ it.effect.each([ error: null, joins: true, }, - // The Jarvis inbox heartbeat failed this way on its first launch and still read as succeeded. { - name: "fails workspace preparation on its first launch", - run: "failed", + name: "loses its soft-deleted thread", + run: "completed", status: "failed", - error: PREPARATION_FAILURE, - unbound: true, + error: THREAD_DELETED, + softDeleted: true, + }, + { + name: "loses its thread while it runs", + run: "completed", + status: "failed", + error: THREAD_DELETED, + waitFails: "thread", + }, + { + name: "is removed while it runs", + run: "completed", + status: "failed", + error: "The run was removed before it finished.", + waitFails: "run", }, ] as const)("a run that $name after its message was accepted records its outcome", (outcome) => Effect.gen(function* () { @@ -454,65 +501,75 @@ it.effect.each([ const ended = yield* Deferred.make(); const sent = yield* Ref.make(""); const run = { id: "run:2", status: outcome.run, rootNodeId: "node:root" }; - const layer = ScheduledTaskService.layer.pipe( - Layer.provide( - Layer.mergeAll( - ManagedPolicy.layerTest(), - Layer.mock(ThreadLaunchService.ThreadLaunchService)({ - launch: (input) => - Ref.set(sent, input.initialMessage?.messageId ?? "").pipe( - Effect.as({ threadId } as never), - ), - }), - 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 }, - ], - messages: [{ id: userMessageId, runId: joins ? "run:3" : "run:2" }], - turnItems: [ - { - type: "error", - status: "failed", - runId: "run:2", - nodeId: "node:root", - failure: { - message: unbound ? PREPARATION_FAILURE : "Weekly usage limit reached", - }, - }, - ], - }) as never, - ), - waitForThread: (input) => - Deferred.await(ended).pipe( - Effect.as({ - threadId: input.threadId, - run: { - ...run, - id: input.runId, - status: joins && input.runId === "run:3" ? "completed" : run.status, - userMessageId: "", + const layer = followerLayer( + Layer.mock(ThreadManagementService.ThreadManagementService)({ + sendToThread: (input) => Ref.set(sent, input.messageId).pipe(Effect.as({} as never)), + // Deleted once its message was sent, so saving and dispatching still see the thread. + getThreadShell: () => + Effect.map(Ref.get(sent), (messageId) => + threadShell( + projectId, + "softDeleted" in outcome && messageId !== "" + ? { deletedAt: "2026-11-12T18:00:00.000Z" } + : {}, + ), + ), + getProjectThreadRecords: () => + Effect.map( + Ref.get(sent), + (userMessageId) => + ({ + runs: [ + { id: "run:1", status: "completed", userMessageId: "message:earlier" }, + { ...run, userMessageId }, + ], + messages: [{ id: userMessageId, runId: joins ? "run:3" : "run:2" }], + turnItems: [ + { + type: "error", + status: "failed", + runId: "run:2", + nodeId: "node:root", + failure: { + message: unbound ? PREPARATION_FAILURE : "Weekly usage limit reached", + }, }, - timedOut: false, - } as never), - ), - }), - Layer.mock(SecretRequests.SecretRequests)({}), - Layer.mock(ManagedProjectFolders.ManagedProjectFolders)({ - namedProjectsRoot: "/projects", - isScratchProject: () => Effect.succeed(false), + ], + }) as never, + ), + waitForThread: (input) => + Effect.gen(function* () { + yield* Deferred.await(ended); + if ("waitFails" in outcome && outcome.waitFails === "run") { + return yield* new ThreadManagementService.ThreadManagementRunNotFoundError({ + threadId: input.threadId, + runId: input.runId!, + }); + } + if ("waitFails" in outcome) { + return yield* new ThreadManagementService.ThreadManagementThreadNotFoundError({ + projectId, + threadId: input.threadId, + }); + } + return { + threadId: input.threadId, + run: { + ...run, + id: input.runId, + status: joins && input.runId === "run:3" ? "completed" : run.status, + userMessageId: "", + }, + timedOut: false, + } as never; }), - NodeCrypto.layer, - Scheduler.layer, - ), - ), + }), + Layer.mock(ThreadLaunchService.ThreadLaunchService)({ + launch: (input) => + Ref.set(sent, input.initialMessage?.messageId ?? "").pipe( + Effect.as({ threadId } as never), + ), + }), ); yield* Effect.gen(function* () { const service = yield* ScheduledTaskService.ScheduledTaskService; @@ -530,89 +587,189 @@ it.effect.each([ interactionMode: "default", }); const dispatched = yield* service.runNow({ id: task.id }); - assert.equal(dispatched.task.lastRunStatus, "started"); + // Pending: the stored status stays the one before this run, a status every client knows. + assert.equal(dispatched.task.lastRunStatus, "never"); + assert.equal(dispatched.task.runInProgressSince, dispatched.task.lastRunAt); const settled = yield* service.subscribeList().pipe( Stream.map(({ tasks }) => tasks.find((candidate) => candidate.id === task.id)), - Stream.filter((current) => current?.lastRunStatus === outcome.status), + Stream.filter((current) => current?.runInProgressSince === undefined), Stream.runHead, Effect.forkChild, ); yield* Deferred.succeed(ended, undefined); const recorded = Option.getOrUndefined(yield* Fiber.join(settled)); - assert.equal(recorded?.lastRunError, outcome.error); + assert.deepEqual( + { status: recorded?.lastRunStatus, error: recorded?.lastRunError }, + { status: outcome.status, error: outcome.error }, + ); assert.equal(recorded?.runCount, 1); }).pipe(Effect.provide(layer)); }).pipe(Effect.provide(SqlitePersistence.layerMemory)), ); -it.effect("a restart picks up each started run and records how it ended", () => +it.effect("a run still going when the next one starts cannot overwrite the newer run", () => + Effect.gen(function* () { + const projectId = ProjectId.make("project:test"); + const sent = yield* Ref.make>([]); + const endFirst = yield* Deferred.make(); + const endSecond = yield* Deferred.make(); + const layer = followerLayer( + Layer.mock(ThreadManagementService.ThreadManagementService)({ + sendToThread: (input) => + Ref.update(sent, (all) => [...all, input.messageId]).pipe(Effect.as({} as never)), + getThreadShell: () => Effect.succeed(threadShell(projectId)), + getProjectThreadRecords: () => + Effect.map( + Ref.get(sent), + (messages) => + ({ + runs: messages.map((userMessageId, index) => ({ + id: `run:${index}`, + status: "running", + rootNodeId: "node:root", + userMessageId, + })), + messages: messages.map((id, index) => ({ id, runId: `run:${index}` })), + turnItems: [], + }) as never, + ), + waitForThread: (input) => + Deferred.await(input.runId === "run:0" ? endFirst : endSecond).pipe( + Effect.as({ + threadId: input.threadId, + run: { + id: input.runId, + status: input.runId === "run:0" ? "failed" : "completed", + rootNodeId: "node:root", + }, + timedOut: false, + } as never), + ), + }), + ); + yield* Effect.gen(function* () { + const service = yield* ScheduledTaskService.ScheduledTaskService; + const { task } = yield* service.upsert({ + id: ScheduledTaskId.make("task:overlap"), + title: "Hourly sync", + prompt: "Sync upstream", + enabled: true, + schedule: { type: "interval", everyMs: 60 * 60_000 }, + projectId, + threadId: ThreadId.make("thread:overlap"), + workspaceStrategy: { type: "root" }, + modelSelection: { instanceId: ProviderInstanceId.make("codex"), model: "gpt-5.4" }, + runtimeMode: "full-access", + interactionMode: "default", + }); + yield* service.runNow({ id: task.id }); + yield* TestClock.adjust("1 minute"); + const second = yield* service.runNow({ id: task.id }); + const states = yield* Queue.unbounded(); + yield* service.subscribeList().pipe( + Stream.runForEach(({ tasks }) => + Queue.offer( + states, + tasks.find((candidate) => candidate.id === task.id), + ), + ), + Effect.forkChild, + ); + yield* Queue.take(states); + yield* Deferred.succeed(endFirst, undefined); + // The first run's follower notifies when it ends; the second run is still the pending one. + const afterFirst = yield* Queue.take(states); + assert.deepEqual( + { status: afterFirst?.lastRunStatus, since: afterFirst?.runInProgressSince }, + { status: "never", since: second.task.lastRunAt }, + ); + yield* Deferred.succeed(endSecond, undefined); + const afterSecond = yield* Queue.take(states); + assert.deepEqual( + { + status: afterSecond?.lastRunStatus, + error: afterSecond?.lastRunError, + since: afterSecond?.runInProgressSince, + }, + { status: "succeeded", error: null, since: undefined }, + ); + }).pipe(Effect.provide(layer)); + }).pipe(Effect.provide(SqlitePersistence.layerMemory)), +); + +it.effect("a restart finds manual and webhook runs again and records how they ended", () => Effect.gen(function* () { const sql = yield* SqlClient.SqlClient; const startedAt = "2026-11-12T17:44:47.869Z"; const threadId = ThreadId.make("thread:resumed"); - const messageId = `scheduled-task-message:task:resumed:${Date.parse(startedAt)}:scheduled`; - for (const id of ["task:resumed", "task:lost"]) { - yield* insertRow(sql, { id, next: null, enabled: 1, status: "started" }, startedAt); - yield* sql`UPDATE scheduled_tasks SET last_run_at = ${startedAt} WHERE task_id = ${id}`; - } - yield* sql` - INSERT INTO orchestration_v2_projection_messages - (message_id, thread_id, run_id, role, streaming, created_at, updated_at, payload_json) - VALUES (${messageId}, ${threadId}, 'run:1', 'user', 0, ${startedAt}, ${startedAt}, '{}') - `; - const run = { - id: "run:1", - status: "failed", - rootNodeId: "node:root", - userMessageId: messageId, + const messages = { + "task:manual": `scheduled-task-message:task:manual:${Date.parse(startedAt)}:manual`, + "task:webhook": "scheduled-task-message:task:webhook:webhook:delivery-1", }; - const layer = ScheduledTaskService.layer.pipe( - Layer.provide( - Layer.mergeAll( - ManagedPolicy.layerTest(), - Layer.mock(ThreadLaunchService.ThreadLaunchService)({}), - Layer.mock(ThreadManagementService.ThreadManagementService)({ - getThreadShell: () => Effect.succeed(threadShell(ProjectId.make("project:test"))), - getProjectThreadRecords: () => - Effect.succeed({ - runs: [run], - messages: [{ id: messageId, runId: run.id }], - turnItems: [ - { - type: "error", - status: "failed", - runId: run.id, - nodeId: run.rootNodeId, - failure: { message: PREPARATION_FAILURE }, - }, - ], - } as never), - waitForThread: (input) => - Effect.succeed({ threadId: input.threadId, run, timedOut: false } as never), - }), - Layer.mock(SecretRequests.SecretRequests)({}), - Layer.mock(ManagedProjectFolders.ManagedProjectFolders)({ - namedProjectsRoot: "/projects", - isScratchProject: () => Effect.succeed(false), - }), - NodeCrypto.layer, - Scheduler.layer, - ), - ), + for (const id of ["task:manual", "task:webhook", "task:lost"]) { + yield* insertRow(sql, { id, next: null, enabled: 1, status: "succeeded" }, startedAt); + yield* sql` + UPDATE scheduled_tasks + SET last_run_at = ${startedAt}, + last_run_webhook_delivery_id = ${id === "task:webhook" ? "delivery-1" : null} + WHERE task_id = ${id} + `; + } + for (const messageId of Object.values(messages)) { + yield* sql` + INSERT INTO orchestration_v2_projection_messages + (message_id, thread_id, run_id, role, streaming, created_at, updated_at, payload_json) + VALUES (${messageId}, ${threadId}, ${`run:${messageId}`}, 'user', 0, ${startedAt}, ${startedAt}, '{}') + `; + } + const layer = followerLayer( + Layer.mock(ThreadManagementService.ThreadManagementService)({ + getThreadShell: () => Effect.succeed(threadShell(ProjectId.make("project:test"))), + getProjectThreadRecords: () => + Effect.succeed({ + runs: Object.values(messages).map((userMessageId) => ({ + id: `run:${userMessageId}`, + status: "failed", + rootNodeId: "node:root", + userMessageId, + })), + messages: Object.values(messages).map((id) => ({ id, runId: `run:${id}` })), + turnItems: Object.values(messages).map((id) => ({ + type: "error", + status: "failed", + runId: `run:${id}`, + nodeId: "node:root", + failure: { message: PREPARATION_FAILURE }, + })), + } as never), + waitForThread: (input) => + Effect.succeed({ + threadId: input.threadId, + run: { id: input.runId, status: "failed", rootNodeId: "node:root" }, + timedOut: false, + } as never), + }), ); yield* Effect.gen(function* () { const service = yield* ScheduledTaskService.ScheduledTaskService; const settled = yield* service.subscribeList().pipe( Stream.map(({ tasks }) => tasks), - Stream.filter((tasks) => tasks.every((task) => task.lastRunStatus === "failed")), + Stream.filter((tasks) => + tasks.every((task) => task.id === "task:lost" || task.lastRunStatus === "failed"), + ), Stream.runHead, ); - const errors = Object.fromEntries( - Option.getOrThrow(settled).map((task) => [task.id, task.lastRunError]), + const recorded = Object.fromEntries( + Option.getOrThrow(settled).map((task) => [ + task.id, + [task.lastRunStatus, task.lastRunError], + ]), ); - assert.deepEqual(errors, { - "task:resumed": PREPARATION_FAILURE, - "task:lost": "The run's thread was deleted before the run finished.", + assert.deepEqual(recorded, { + "task:manual": ["failed", PREPARATION_FAILURE], + "task:webhook": ["failed", PREPARATION_FAILURE], + // No message to follow: left as it was. + "task:lost": ["succeeded", null], }); }).pipe(Effect.provide(layer)); }).pipe(Effect.provide(SqlitePersistence.layerMemory)), @@ -620,18 +777,33 @@ it.effect("a restart picks up each started run and records how it ended", () => // #69: a task MCP saved to run in its folder because the folder was not a repository. it.effect.each([ - { name: "refuses an unbound MCP task", git: true, dispatch: "refused" }, - { name: "runs a bound task", git: true, bound: true, dispatch: "sent" }, - { name: "runs a task whose folder was chosen in Settings", git: true, source: "web" }, - { name: "runs a task in a scratch project", git: true, scratch: true }, - { name: "runs a task whose folder is still not a repository", git: false }, + { name: "refuses an unbound MCP task", git: "folder", dispatch: "refused" }, + { + name: "refuses an MCP task whose parent folder became a repository", + git: "parent", + dispatch: "refused", + }, + { + name: "refuses an MCP task later edited in Settings", + git: "folder", + editedInSettings: true, + dispatch: "refused", + }, + { name: "runs a bound task", git: "folder", bound: true, dispatch: "sent" }, + { name: "runs a task whose folder was chosen in Settings", git: "folder", source: "web" }, + { name: "runs a task in a scratch project", git: "folder", scratch: true }, + { name: "runs a task whose folder is still not a repository", git: "none" }, ] as const)("a saved project-folder workspace in a folder that became a repository $name", (row) => Effect.gen(function* () { const fileSystem = yield* FileSystem.FileSystem; const path = yield* Path.Path; const sql = yield* SqlClient.SqlClient; - const root = yield* fileSystem.makeTempDirectoryScoped({ prefix: "scheduled-root-" }); - if (row.git) yield* fileSystem.makeDirectory(path.join(root, ".git")); + const parent = yield* fileSystem.makeTempDirectoryScoped({ prefix: "scheduled-root-" }); + const root = path.join(parent, "project"); + yield* fileSystem.makeDirectory(root); + if (row.git !== "none") { + yield* fileSystem.makeDirectory(path.join(row.git === "parent" ? parent : root, ".git")); + } const projectId = ProjectId.make("project:repo"); yield* sql` INSERT INTO projection_projects (project_id, title, workspace_root, scripts_json, created_at, updated_at) @@ -666,7 +838,7 @@ it.effect.each([ ); yield* Effect.gen(function* () { const service = yield* ScheduledTaskService.ScheduledTaskService; - const { task } = yield* service.upsert({ + const input: ScheduledTaskUpsertInput = { id: ScheduledTaskId.make("task:repo"), title: "Nightly notes", prompt: "Update the notes", @@ -680,15 +852,25 @@ it.effect.each([ interactionMode: "default", createdBy: "agent", creationSource: "source" in row ? row.source : "mcp", - }); + }; + const { task } = yield* service.upsert(input); + if ("editedInSettings" in row) { + const edited = yield* service.upsert({ ...input, title: "Renamed", creationSource: "web" }); + assert.equal(edited.task.creationSource, "mcp"); + } const result = yield* service.runNow({ id: task.id }); assert.equal(yield* Ref.get(dispatched), "dispatch" in row ? row.dispatch : "launched"); if ("dispatch" in row && row.dispatch === "refused") { assert.equal(result.task.lastRunStatus, "failed"); - assert.include(result.task.lastRunError, `${root} is now a Git repository`); assert.equal(result.task.runCount, 1); + assert.include( + result.task.lastRunError, + row.git === "parent" + ? `inside the Git repository at ${parent} (${path.join(parent, ".git")})` + : `${root} is now a Git repository (${path.join(root, ".git")})`, + ); } else { - assert.equal(result.task.lastRunStatus, "started"); + assert.equal(result.task.runInProgressSince, result.task.lastRunAt); } }).pipe(Effect.provide(layer)); }).pipe(Effect.provide(Layer.merge(SqlitePersistence.layerMemory, NodeServices.layer))), diff --git a/apps/server/src/scheduledTasks/ScheduledTaskService.ts b/apps/server/src/scheduledTasks/ScheduledTaskService.ts index 4c4c3a310c686..2cfed6f3f5c60 100644 --- a/apps/server/src/scheduledTasks/ScheduledTaskService.ts +++ b/apps/server/src/scheduledTasks/ScheduledTaskService.ts @@ -1,6 +1,9 @@ import { CommandId, MessageId, + type OrchestrationV2Run, + OrchestrationV2RunStatus, + type RunId, PROVIDER_SEND_TURN_MAX_INPUT_CHARS, type ProjectId, ScheduledTask, @@ -15,6 +18,7 @@ import { type ScheduledTaskListWebhookDeliveriesInput, type ScheduledTaskListWebhookDeliveriesResult, type ScheduledTaskRotateWebhookTokenInput, + type ScheduledTaskRunStatus, type ScheduledTaskWebhookDeliveryOutcome, type ScheduledTaskWebhookFilter, type ScheduledTaskListResult, @@ -59,7 +63,7 @@ import * as DueHint from "../scheduling/DueHint.ts"; import * as ScheduledTaskChanges from "./ScheduledTaskChanges.ts"; import * as Scheduler from "../scheduling/Scheduler.ts"; import { latestRootProviderFailure } from "@t3tools/shared/orchestrationV2ThreadError"; -import { hasGitEntry } from "../mcp/gitEntryProbe.ts"; +import { findGitEntry } from "../mcp/gitEntryProbe.ts"; import { isMissedFixedTimeRun, isSameSchedule, nextScheduledRunAt } from "./Schedule.ts"; import { redactHeaders, @@ -235,9 +239,10 @@ interface ScheduledTaskRow { readonly webhook_secret: string | null; } -interface StartedRunRow { +interface LastRunRow { readonly task_id: string; readonly last_run_at: string; + readonly last_run_status: string; readonly last_run_webhook_delivery_id: string | null; } @@ -322,6 +327,12 @@ const UNSUCCESSFUL_RUN_ERRORS = { cancelled: "The run was cancelled.", rolled_back: "The run was rolled back.", } as const; +type RunOutcome = { readonly status: "succeeded" | "failed"; readonly error: string | null }; +const THREAD_DELETED_OUTCOME: RunOutcome = { + status: "failed", + error: "The run's thread was deleted before the run finished.", +}; +const isRunStatus = Schema.is(OrchestrationV2RunStatus); const localNow = DateTime.withCurrentZoneLocal(DateTime.nowInCurrentZone); function nextRunAt( @@ -487,10 +498,22 @@ export const layer = Layer.effect( webhookPolicyOrganization === undefined ? undefined : managedPolicyDeniedMessage(webhookPolicyOrganization, "webhook tasks"); - const withPolicy = (task: ScheduledTask): ScheduledTask => - webhookPolicyBlock !== undefined && task.schedule.type === "webhook" - ? { ...task, blockedByPolicy: webhookPolicyBlock } - : task; + // Accepted runs whose outcome is pending, by task, with the run's start time. Kept out of the + // table so stored statuses stay ones every client and older build can decode. + const pendingRuns = yield* Ref.make>(new Map()); + const present = ( + task: ScheduledTask, + pending: ReadonlyMap, + ): ScheduledTask => { + const since = pending.get(task.id); + return { + ...task, + ...(since !== undefined && since === task.lastRunAt ? { runInProgressSince: since } : {}), + ...(webhookPolicyBlock !== undefined && task.schedule.type === "webhook" + ? { blockedByPolicy: webhookPolicyBlock } + : {}), + }; + }; const webhookPermits = yield* Ref.make>( new Map(), ); @@ -552,9 +575,10 @@ export const layer = Layer.effect( const listRows = Effect.fn("ScheduledTaskService.listRows")(function* () { const rows = yield* selectAllRows(); const origin = yield* readWebhookOrigin; + const pending = yield* Ref.get(pendingRuns); return yield* Effect.forEach( rows, - (row) => decodeRow(row, origin).pipe(Effect.map(withPolicy)), + (row) => decodeRow(row, origin).pipe(Effect.map((task) => present(task, pending))), { concurrency: 1 }, ); }); @@ -597,7 +621,7 @@ export const layer = Layer.effect( ); const row = rows[0]; if (row === undefined) return null; - return withPolicy(yield* decodeRow(row, yield* readWebhookOrigin)); + return present(yield* decodeRow(row, yield* readWebhookOrigin), yield* Ref.get(pendingRuns)); }); const findWebhookCredentials = (id: ScheduledTaskId) => @@ -784,7 +808,7 @@ export const layer = Layer.effect( readonly id: ScheduledTaskId; readonly completedAtIso: string; readonly nextRunAtIso: string | null; - readonly status: "started" | "failed"; + readonly status: Exclude; readonly error: string | null; readonly startedAtIso: string; }) => @@ -887,58 +911,58 @@ export const layer = Layer.effect( ), ); - const recordRunOutcome = ( + const setPendingRun = (taskId: ScheduledTaskId, startedAtIso: string) => + Ref.update(pendingRuns, (pending) => new Map(pending).set(taskId, startedAtIso)); + const clearPendingRun = (taskId: ScheduledTaskId, startedAtIso?: string) => + Ref.modify(pendingRuns, (pending) => { + const current = pending.get(taskId); + if (current === undefined || (startedAtIso !== undefined && current !== startedAtIso)) + return [false, pending] as const; + const next = new Map(pending); + next.delete(taskId); + return [true, next] as const; + }); + + // Only while the run is still the pending one, so a missed-run failure recorded meanwhile stands. + const recordRunOutcome = Effect.fn("ScheduledTaskService.recordRunOutcome")(function* ( taskId: ScheduledTaskId, startedAtIso: string, - status: "succeeded" | "failed", - error: string | null, - ) => - sql` + outcome: RunOutcome, + ) { + if ((yield* Ref.get(pendingRuns)).get(taskId) !== startedAtIso) return; + yield* sql` UPDATE scheduled_tasks - SET last_run_status = ${status}, last_run_error = ${error} - WHERE task_id = ${taskId} AND last_run_at = ${startedAtIso} AND last_run_status = 'started' - RETURNING task_id - `.pipe(Effect.flatMap((rows) => (rows.length > 0 ? notifyChanged : Effect.void))); + SET last_run_status = ${outcome.status}, last_run_error = ${outcome.error} + WHERE task_id = ${taskId} AND last_run_at = ${startedAtIso} + AND last_run_status <> 'running' + `; + }); - // A dispatch is recorded as started once the message is accepted; workspace preparation - // or the provider can still fail the run after that, so follow it to its outcome. - const followRunOutcome = Effect.fn("ScheduledTaskService.followRunOutcome")( - function* ( - taskId: ScheduledTaskId, - startedAtIso: string, - threadId: ThreadId, - messageId: MessageId, - ) { + // Null while the run is still active after the wait. + const resolveRunOutcome = Effect.fn("ScheduledTaskService.resolveRunOutcome")( + function* (threadId: ThreadId, messageId: MessageId) { const shell = yield* threadManagement.getThreadShell(threadId); - if (shell === null) { - return yield* recordRunOutcome( - taskId, - startedAtIso, - "failed", - "The run's thread was deleted before the run finished.", - ); - } + if (shell === null || shell.deletedAt != null) return THREAD_DELETED_OUTCOME; const thread = { projectId: shell.projectId, threadId }; const { runs, messages } = yield* threadManagement.getProjectThreadRecords(thread, [ "runs", "messages", ]); - let runId = + let runId: RunId | null = runs.find((run) => run.userMessageId === messageId)?.id ?? messages.find((message) => message.id === messageId)?.runId ?? null; + if (runId === null) return { status: "failed", error: "The run never started." } as const; // A queued message promoted to a steer cancels its run and joins the active one. - for (let hops = 0; runId !== null && hops < 3; hops++) { - const { run } = yield* threadManagement.waitForThread({ - ...thread, - runId, - timeoutMs: RUN_OUTCOME_WAIT_MS, - }); - // Still active after the wait: the run has not finished, so it stays started. - if (run === null || !ThreadManagementService.isTerminalRunStatus(run.status)) return; - if (run.status === "completed") { - return yield* recordRunOutcome(taskId, startedAtIso, "succeeded", null); - } + for (let hops = 0; hops < 3; hops++) { + const { run }: { readonly run: OrchestrationV2Run | null } = + yield* threadManagement.waitForThread({ + ...thread, + runId, + timeoutMs: RUN_OUTCOME_WAIT_MS, + }); + if (run === null || !ThreadManagementService.isTerminalRunStatus(run.status)) return null; + if (run.status === "completed") return { status: "succeeded", error: null } as const; const { turnItems, messages: current } = yield* threadManagement.getProjectThreadRecords( thread, ["turnItems", "messages"], @@ -946,59 +970,83 @@ export const layer = Layer.effect( const joined = current.find((message) => message.id === messageId)?.runId ?? null; // waitForThread answers with a restart's continuation, so compare against both ids. if ( - run.status === "cancelled" && - joined !== null && - joined !== runId && - joined !== run.id + run.status !== "cancelled" || + joined === null || + joined === runId || + joined === run.id ) { - runId = joined; - continue; + const error = + latestRootProviderFailure(run, turnItems)?.message ?? + UNSUCCESSFUL_RUN_ERRORS[run.status]; + return { status: "failed", error } as const; } - const error = - latestRootProviderFailure(run, turnItems)?.message ?? - UNSUCCESSFUL_RUN_ERRORS[run.status]; - return yield* recordRunOutcome(taskId, startedAtIso, "failed", error); + runId = joined; } - yield* recordRunOutcome( - taskId, - startedAtIso, - "failed", - runId === null ? "The run never started." : "The run was cancelled.", - ); + return { status: "failed", error: UNSUCCESSFUL_RUN_ERRORS.cancelled } as const; }, - (effect, taskId) => - effect.pipe( - Effect.catchCause((cause) => - Effect.logWarning("Could not record scheduled run outcome", { taskId, cause }), - ), - ), + Effect.catchTags({ + ThreadManagementThreadNotFoundError: () => Effect.succeed(THREAD_DELETED_OUTCOME), + ThreadManagementRunNotFoundError: () => + Effect.succeed({ + status: "failed", + error: "The run was removed before it finished.", + } as const), + }), ); - // The message id is derived from the run's start time and trigger (see runTask's fireKey), - // so a restart can find a started run again and keep following it. - const resumeStartedRun = Effect.fn("ScheduledTaskService.resumeStartedRun")(function* ( - row: StartedRunRow, + // The caller marks the run pending first. Any other failure leaves the stored status as it was. + const followRunOutcome = ( + taskId: ScheduledTaskId, + startedAtIso: string, + threadId: ThreadId, + messageId: MessageId, + ) => + resolveRunOutcome(threadId, messageId).pipe( + Effect.flatMap((outcome) => + outcome === null ? Effect.void : recordRunOutcome(taskId, startedAtIso, outcome), + ), + Effect.catchCause((cause) => + Effect.logWarning("Could not record scheduled run outcome", { taskId, cause }), + ), + // Notifies even when a newer run replaced this one, so every follower's end is observable. + Effect.ensuring(clearPendingRun(taskId, startedAtIso).pipe(Effect.andThen(notifyChanged))), + ); + + // The stored row holds no trace of a pending run, so after a restart each task's latest run is + // found again from its message id (see runTask's fireKey) and followed when its outcome is not + // recorded yet. A row whose message is gone is left as it is. + const resumeRunOutcome = Effect.fn("ScheduledTaskService.resumeRunOutcome")(function* ( + row: LastRunRow, ) { - const taskId = ScheduledTaskId.make(row.task_id); const startedMs = Date.parse(row.last_run_at); const fireKeys = row.last_run_webhook_delivery_id !== null ? [`${row.task_id}:webhook:${row.last_run_webhook_delivery_id}`] : [`${row.task_id}:${startedMs}:scheduled`, `${row.task_id}:${startedMs}:manual`]; - const messageIds = fireKeys.map((key) => `scheduled-task-message:${key}`); - const [found] = yield* sql<{ message_id: string; thread_id: string }>` - SELECT message_id, thread_id FROM orchestration_v2_projection_messages - WHERE ${sql.in("message_id", messageIds)} + const [found] = yield* sql<{ + message_id: string; + thread_id: string; + run_status: string | null; + }>` + SELECT m.message_id, m.thread_id, r.status AS run_status + FROM orchestration_v2_projection_messages m + LEFT JOIN orchestration_v2_projection_runs r ON r.run_id = m.run_id + WHERE ${sql.in( + "m.message_id", + fireKeys.map((key) => `scheduled-task-message:${key}`), + )} LIMIT 1 `.pipe(NodeSqliteClient.readOnly); - if (found === undefined) { - return yield* recordRunOutcome( - taskId, - row.last_run_at, - "failed", - "The run's thread was deleted before the run finished.", - ); + if (found === undefined) return; + if ( + isRunStatus(found.run_status) && + ThreadManagementService.isTerminalRunStatus(found.run_status) + ) { + const recorded = found.run_status === "completed" ? "succeeded" : "failed"; + if (row.last_run_status === recorded) return; } + const taskId = ScheduledTaskId.make(row.task_id); + yield* setPendingRun(taskId, row.last_run_at); yield* followRunOutcome( taskId, row.last_run_at, @@ -1022,9 +1070,13 @@ export const layer = Layer.effect( const [project] = yield* sql<{ workspace_root: string }>` SELECT workspace_root FROM projection_projects WHERE project_id = ${task.projectId} `.pipe(NodeSqliteClient.readOnly); - if (project === undefined || !(yield* hasGitEntry(project.workspace_root))) return; + if (project === undefined) return; + const gitEntry = yield* findGitEntry(project.workspace_root); + if (gitEntry === null) return; return yield* taskError( - `${project.workspace_root} is now a Git repository, but this task was saved to run directly in that folder because it was not one. Update the task (any change through schedule_task re-checks the folder) or choose a workspace in Settings.`, + gitEntry.inParent + ? `${project.workspace_root} is now inside the Git repository at ${gitEntry.folder} (${gitEntry.path}), but this task was saved to run directly in that folder because it was not in one. Updating the task through schedule_task would run it in a worktree of ${gitEntry.folder}; if that is not intended, choose a workspace in Settings.` + : `${project.workspace_root} is now a Git repository (${gitEntry.path}), but this task was saved to run directly in that folder because it was not one. Update the task (any change through schedule_task re-checks the folder) or choose a workspace in Settings.`, { taskId: task.id }, ); }, @@ -1209,8 +1261,13 @@ export const layer = Layer.effect( const runSucceeded = result._tag === "Success"; if (webhook !== undefined && runSucceeded) yield* Ref.set(webhook.dispatched, true); const completedAt = yield* localNow; - const lastRunStatus = runSucceeded ? ("started" as const) : ("failed" as const); - const lastRunError = runSucceeded ? null : errorMessage(result.cause); + // An accepted run keeps the previous run's status until its own outcome is known. + const lastRunStatus = !runSucceeded + ? ("failed" as const) + : active.lastRunStatus === "running" + ? ("never" as const) + : active.lastRunStatus; + const lastRunError = runSucceeded ? active.lastRunError : errorMessage(result.cause); const threadId = runSucceeded ? result.value : null; // One transaction, so an edit cannot land between reading the schedule and // writing the next run computed from it. The origin is read first: it reads @@ -1275,14 +1332,20 @@ export const layer = Layer.effect( ), ), ); + const following = written && threadId !== null; + if (following) yield* setPendingRun(active.id, startedAtIso); if (written) yield* notifyChanged; - if (written && threadId !== null) { + if (following) { yield* followRunOutcome(active.id, startedAtIso, threadId, messageId).pipe( Effect.forkIn(serviceScope), ); } // The stored row can belong to a replacement, so report this dispatch separately. - return { task: recorded, threadId, dispatchFailed: !runSucceeded }; + return { + task: following ? { ...recorded, runInProgressSince: startedAtIso } : recorded, + threadId, + dispatchFailed: !runSucceeded, + }; }).pipe( Effect.onError((cause) => releaseStuckRun(task, startedAtIso, errorMessage(cause), webhook?.deliveryId), @@ -1342,7 +1405,10 @@ export const layer = Layer.effect( taskError("Could not reschedule missed schedule task run.", { taskId: task.id, cause }), ), ); - if (updated.length > 0) yield* notifyChanged; + if (updated.length === 0) return; + // The recorded miss stands over the outcome of a run still pending from before it. + if (runError !== undefined) yield* clearPendingRun(task.id); + yield* notifyChanged; }); const runDueTasks = Effect.fn("ScheduledTaskService.runDueTasks")(function* () { @@ -1396,6 +1462,33 @@ export const layer = Layer.effect( ); }); + // Before the running-row recovery below, whose rows never dispatched a message to follow. + yield* sql` + SELECT task_id, last_run_at, last_run_status, last_run_webhook_delivery_id + FROM scheduled_tasks + WHERE last_run_at IS NOT NULL AND last_run_status <> 'running' + `.pipe( + NodeSqliteClient.readOnly, + Effect.flatMap((rows) => + Effect.forEach( + rows, + (row) => + resumeRunOutcome(row).pipe( + Effect.catch((cause) => + Effect.logWarning("Could not resume following a schedule task run", { + taskId: row.task_id, + cause, + }), + ), + ), + { discard: true }, + ), + ), + Effect.catch((cause) => + Effect.logWarning("Could not resume following schedule task runs", { cause }), + ), + ); + // Recover crash-left running rows that the due-task filter would otherwise skip forever. yield* Effect.gen(function* () { const rows = yield* selectAllRows(); @@ -1449,32 +1542,6 @@ export const layer = Layer.effect( ), ); - // A run's outcome follower does not survive a restart; pick each started run up again. - yield* sql` - SELECT task_id, last_run_at, last_run_webhook_delivery_id FROM scheduled_tasks - WHERE last_run_status = 'started' AND last_run_at IS NOT NULL - `.pipe( - NodeSqliteClient.readOnly, - Effect.flatMap((rows) => - Effect.forEach( - rows, - (row) => - resumeStartedRun(row).pipe( - Effect.catch((cause) => - Effect.logWarning("Could not resume following a started schedule task run", { - taskId: row.task_id, - cause, - }), - ), - ), - { discard: true }, - ), - ), - Effect.catch((cause) => - Effect.logWarning("Could not resume following started schedule task runs", { cause }), - ), - ); - yield* scheduler.register("scheduled-tasks", runDueTasks()); const list: ScheduledTaskService["Service"]["list"] = () => @@ -1685,7 +1752,8 @@ export const layer = Layer.effect( runtimeMode: input.runtimeMode, interactionMode: input.interactionMode, createdBy: existingTask?.createdBy ?? input.createdBy ?? "user", - creationSource: input.creationSource ?? "web", + // Kept on edits: MCP's derived workspaces are rechecked at launch (refuseStaleRootWorkspace). + creationSource: existingTask?.creationSource ?? input.creationSource ?? "web", createdAt: existingTask?.createdAt ?? iso(now), updatedAt: iso(now), nextRunAt: scheduleUnchanged diff --git a/apps/server/src/scheduledTasks/ScheduledTaskService.webhook.test.ts b/apps/server/src/scheduledTasks/ScheduledTaskService.webhook.test.ts index a216e3a91923e..75b9c5eebb80d 100644 --- a/apps/server/src/scheduledTasks/ScheduledTaskService.webhook.test.ts +++ b/apps/server/src/scheduledTasks/ScheduledTaskService.webhook.test.ts @@ -1082,7 +1082,7 @@ it.effect("retrying a running window keeps the prompt it was dispatched with", ( yield* service.triggerWebhook(requestFor(task)); yield* save("Prompt at dispatch"); yield* sql`CREATE TRIGGER fail_completion BEFORE UPDATE ON scheduled_tasks - WHEN NEW.last_run_status = 'started' BEGIN SELECT RAISE(FAIL, 'completion unavailable'); END`; + WHEN OLD.last_run_status = 'running' AND NEW.last_run_status <> 'failed' BEGIN SELECT RAISE(FAIL, 'completion unavailable'); END`; yield* TestClock.adjust("10 seconds"); yield* drainPendingDispatch(workers); const launched = yield* Queue.take(launches); @@ -1408,7 +1408,8 @@ it.effect("start failures release both replay keys and relay claims for redelive ); assert.equal(retry._tag === "accepted" && retry.outcome, "accepted"); yield* Queue.take(launches); - assert.equal((yield* service.list()).tasks[0]!.lastRunStatus, "started"); + // The retry was accepted; the failed start stays the latest outcome until it finishes. + assert.equal((yield* service.list()).tasks[0]!.runCount, 2); }), { failLaunches: { remaining: 1 } }, ), @@ -1528,7 +1529,7 @@ it.effect("keeps replay keys when dispatch succeeds but completion bookkeeping f }), ); yield* sql`CREATE TRIGGER fail_completion BEFORE UPDATE ON scheduled_tasks - WHEN NEW.last_run_status = 'started' BEGIN SELECT RAISE(FAIL, 'completion unavailable'); END`; + WHEN OLD.last_run_status = 'running' AND NEW.last_run_status <> 'failed' BEGIN SELECT RAISE(FAIL, 'completion unavailable'); END`; yield* waitForSpan( service.triggerWebhook(requestFor(task)), "ScheduledTaskService.runWebhookDelivery", @@ -1920,7 +1921,7 @@ it.effect.each([false, true])( requestFor(task, { headers: { group: "old" }, bodyText: '{"value":"old"}' }), ); yield* sql`CREATE TRIGGER fail_old_completion BEFORE UPDATE ON scheduled_tasks - WHEN NEW.last_run_status = 'started' BEGIN SELECT RAISE(FAIL, 'completion unavailable'); END`; + WHEN OLD.last_run_status = 'running' AND NEW.last_run_status <> 'failed' BEGIN SELECT RAISE(FAIL, 'completion unavailable'); END`; yield* TestClock.adjust("10 seconds"); yield* drainPendingDispatch(workers); yield* Queue.take(launches); diff --git a/apps/web/src/components/chat/ThreadAutomationsPanel.tsx b/apps/web/src/components/chat/ThreadAutomationsPanel.tsx index 93e87a305c45c..6b3ac2f5d67aa 100644 --- a/apps/web/src/components/chat/ThreadAutomationsPanel.tsx +++ b/apps/web/src/components/chat/ThreadAutomationsPanel.tsx @@ -7,6 +7,7 @@ import { AuthOrchestrationOperateScope, type EnvironmentId, type ScheduledTask, + scheduledTaskRunDisplayStatus, type ThreadId, } from "@t3tools/contracts"; import { @@ -34,7 +35,6 @@ import { const STATUS_DOT_CLASS: Record = { never: "bg-muted-foreground/40", running: "bg-sky-500", - started: "bg-sky-500", succeeded: "bg-emerald-500", failed: "bg-destructive", }; @@ -171,7 +171,7 @@ export function ThreadAutomationsPanel(props: { diff --git a/apps/web/src/components/settings/ScheduledTasksSettings.tsx b/apps/web/src/components/settings/ScheduledTasksSettings.tsx index 9eb1dc33a0af2..18791f4c8a6ab 100644 --- a/apps/web/src/components/settings/ScheduledTasksSettings.tsx +++ b/apps/web/src/components/settings/ScheduledTasksSettings.tsx @@ -84,6 +84,7 @@ import { AuthOrchestrationOperateScope, clampManagedRuntimeMode, managedCapabilityDenial, + scheduledTaskRunDisplayStatus, } from "@t3tools/contracts"; import { resolvePastedPath } from "@t3tools/shared/path"; import { useManagedPolicy } from "~/state/managedPolicy"; @@ -364,7 +365,7 @@ function deliveryOutcomeVariant(outcome: string) { function statusVariant(status: ScheduledTask["lastRunStatus"]) { if (status === "failed") return "error"; if (status === "succeeded") return "success"; - if (status === "running" || status === "started") return "info"; + if (status === "running") return "info"; return "outline"; } @@ -741,6 +742,7 @@ function ScheduledTaskRow({ const [busy, setBusy] = useState(false); const [deliveriesOpen, setDeliveriesOpen] = useState(false); const isWebhook = task.schedule.type === "webhook"; + const runStatus = scheduledTaskRunDisplayStatus(task); const toggle = useAtomCommand(serverEnvironment.setScheduledTaskEnabled, { label: "scheduled task enabled", }); @@ -792,8 +794,8 @@ function ScheduledTaskRow({ "Not scheduled" )} - {task.lastRunStatus !== "never" ? ( - {task.lastRunStatus} + {runStatus !== "never" ? ( + {runStatus} ) : null} {task.threadId !== null && scheduledTaskThreadMode(task) !== "new_thread" ? ( diff --git a/packages/contracts/src/orchestratorMcp.ts b/packages/contracts/src/orchestratorMcp.ts index 0b06dc4b12303..1bc807413cdea 100644 --- a/packages/contracts/src/orchestratorMcp.ts +++ b/packages/contracts/src/orchestratorMcp.ts @@ -769,6 +769,20 @@ export const OrchestratorMcpSaveScheduledTaskInput = Schema.Struct({ export type OrchestratorMcpSaveScheduledTaskInput = typeof OrchestratorMcpSaveScheduledTaskInput.Type; +/** `started`: the latest run was accepted and has not finished; the others describe a finished run. */ +export const OrchestratorMcpScheduledTaskRunStatus = Schema.Literals([ + ...ScheduledTaskRunStatus.literals, + "started", +]); +export type OrchestratorMcpScheduledTaskRunStatus = + typeof OrchestratorMcpScheduledTaskRunStatus.Type; + +export const orchestratorMcpScheduledTaskRunStatus = (task: { + readonly lastRunStatus: ScheduledTaskRunStatus; + readonly runInProgressSince?: string; +}): OrchestratorMcpScheduledTaskRunStatus => + task.runInProgressSince === undefined ? task.lastRunStatus : "started"; + export const OrchestratorMcpScheduledTask = Schema.Struct({ scheduledTaskId: ScheduledTaskId, title: Schema.String, @@ -779,7 +793,7 @@ export const OrchestratorMcpScheduledTask = Schema.Struct({ freshSession: Schema.Boolean, schedule: ScheduledTaskSchedule, nextRunAt: Schema.NullOr(IsoDateTime), - lastRunStatus: ScheduledTaskRunStatus, + lastRunStatus: OrchestratorMcpScheduledTaskRunStatus, webhookUrl: Schema.optional(Schema.String).annotate({ description: "Public URL for the sender. Absent: no T3 Connect tunnel. The user must enable T3 Connect remote access.", diff --git a/packages/contracts/src/scheduledTask.ts b/packages/contracts/src/scheduledTask.ts index 1c139aa0a6a8e..5beb4b0672ef5 100644 --- a/packages/contracts/src/scheduledTask.ts +++ b/packages/contracts/src/scheduledTask.ts @@ -214,14 +214,7 @@ export const ScheduledTaskUpsertSchedule = Schema.Union([ ]); export type ScheduledTaskUpsertSchedule = typeof ScheduledTaskUpsertSchedule.Type; -/** `running` while the run is being dispatched; `started` once it was accepted and its outcome is pending. */ -export const ScheduledTaskRunStatus = Schema.Literals([ - "never", - "running", - "started", - "succeeded", - "failed", -]); +export const ScheduledTaskRunStatus = Schema.Literals(["never", "running", "succeeded", "failed"]); export type ScheduledTaskRunStatus = typeof ScheduledTaskRunStatus.Type; export const ScheduledTaskWebhookEndpoint = Schema.Struct({ @@ -270,9 +263,22 @@ export const ScheduledTask = Schema.Struct({ webhook: Schema.optional(ScheduledTaskWebhookEndpoint), /** Why the task cannot run regardless of `enabled`, such as a managed policy refusing webhooks. */ blockedByPolicy: Schema.optionalKey(Schema.String), + /** + * When the run started at `lastRunAt` was accepted and has not finished yet. The `lastRun*` fields + * still describe the run before it until it ends. Never stored, so older clients and builds see + * only the statuses they know. + */ + runInProgressSince: Schema.optionalKey(IsoDateTime), }); export type ScheduledTask = typeof ScheduledTask.Type; +/** What to show for the latest run: `running` until an accepted run's outcome is known. */ +export function scheduledTaskRunDisplayStatus( + task: Pick, +): ScheduledTaskRunStatus { + return task.runInProgressSince === undefined ? task.lastRunStatus : "running"; +} + export const ScheduledTaskListInput = Schema.Struct({}); export type ScheduledTaskListInput = typeof ScheduledTaskListInput.Type; From 480176e0d080ee3b611821c045bee53d49da671f Mon Sep 17 00:00:00 2001 From: Chris OHearn Date: Thu, 12 Nov 2026 14:51:10 -0700 Subject: [PATCH 4/6] ci: re-run checks Co-Authored-By: Claude Opus 5.5 From ffe675ebd38f07a354bb583081055e5000c8114a Mon Sep 17 00:00:00 2001 From: Chris OHearn Date: Thu, 12 Nov 2026 14:54:25 -0700 Subject: [PATCH 5/6] ci: re-run checks with CI enabled Co-Authored-By: Claude Opus 5.5 From a744043f2569f494a4bfdcc9b19b75ce71aba68d Mon Sep 17 00:00:00 2001 From: Chris OHearn Date: Thu, 12 Nov 2026 15:09:39 -0700 Subject: [PATCH 6/6] fix(server): a missed slot becomes the latest run so no older run's outcome replaces it - A recorded miss moves last_run_at to now and clears the webhook delivery id, so a restart no longer finds the earlier run again and records its success. - markRunning reads the status it replaces in its own transaction. - A restart skips only a run already recorded as a clean success. - Web hides the previous error while a run is in progress and disables Run now from the displayed status. Co-Authored-By: Claude Opus 5.5 --- .../ScheduledTaskService.schedule.test.ts | 14 +- .../ScheduledTaskService.test.ts | 53 +++++--- .../scheduledTasks/ScheduledTaskService.ts | 125 +++++++++++------- .../chat/ThreadAutomationsPanel.tsx | 4 +- .../settings/ScheduledTasksSettings.tsx | 4 +- 5 files changed, 130 insertions(+), 70 deletions(-) diff --git a/apps/server/src/scheduledTasks/ScheduledTaskService.schedule.test.ts b/apps/server/src/scheduledTasks/ScheduledTaskService.schedule.test.ts index 3e1bb4d001011..26786628e55b7 100644 --- a/apps/server/src/scheduledTasks/ScheduledTaskService.schedule.test.ts +++ b/apps/server/src/scheduledTasks/ScheduledTaskService.schedule.test.ts @@ -553,7 +553,7 @@ it.effect( // The server is off across the 09:00 slot while the 08:00 run keeps going. yield* TestClock.setTime(DateTime.toEpochMillis(at(9, 30))); - yield* Effect.gen(function* () { + const missedError = yield* Effect.gen(function* () { const service = yield* ScheduledTaskService.ScheduledTaskService; const states = yield* Queue.unbounded(); yield* service.subscribeList().pipe( @@ -576,7 +576,19 @@ it.effect( error: afterRun?.lastRunError, since: afterRun?.runInProgressSince, }).toEqual({ status: "failed", error: missed?.lastRunError, since: undefined }); + return missed?.lastRunError; + }).pipe(Effect.provide(layer)); + + // A second restart must not find the 08:00 run again and record its success over the miss. + const [again] = yield* Effect.gen(function* () { + const service = yield* ScheduledTaskService.ScheduledTaskService; + return (yield* service.list()).tasks; }).pipe(Effect.provide(layer)); + expect({ + status: again?.lastRunStatus, + error: again?.lastRunError, + since: again?.runInProgressSince, + }).toEqual({ status: "failed", error: missedError, since: undefined }); }).pipe(Effect.provide(SqlitePersistence.layerMemory)), ); diff --git a/apps/server/src/scheduledTasks/ScheduledTaskService.test.ts b/apps/server/src/scheduledTasks/ScheduledTaskService.test.ts index 1170e8850a89f..342688b025959 100644 --- a/apps/server/src/scheduledTasks/ScheduledTaskService.test.ts +++ b/apps/server/src/scheduledTasks/ScheduledTaskService.test.ts @@ -697,16 +697,20 @@ it.effect("a run still going when the next one starts cannot overwrite the newer }).pipe(Effect.provide(SqlitePersistence.layerMemory)), ); -it.effect("a restart finds manual and webhook runs again and records how they ended", () => +it.effect("a restart finds each trigger's run again and records how it ended", () => Effect.gen(function* () { const sql = yield* SqlClient.SqlClient; const startedAt = "2026-11-12T17:44:47.869Z"; + const startedMs = Date.parse(startedAt); const threadId = ThreadId.make("thread:resumed"); - const messages = { - "task:manual": `scheduled-task-message:task:manual:${Date.parse(startedAt)}:manual`, + const release = yield* Deferred.make(); + const followed = { + "task:scheduled": `scheduled-task-message:task:scheduled:${startedMs}:scheduled`, + "task:manual": `scheduled-task-message:task:manual:${startedMs}:manual`, "task:webhook": "scheduled-task-message:task:webhook:webhook:delivery-1", }; - for (const id of ["task:manual", "task:webhook", "task:lost"]) { + const matched = `scheduled-task-message:task:matched:${startedMs}:scheduled`; + for (const id of [...Object.keys(followed), "task:matched", "task:lost"]) { yield* insertRow(sql, { id, next: null, enabled: 1, status: "succeeded" }, startedAt); yield* sql` UPDATE scheduled_tasks @@ -715,26 +719,32 @@ it.effect("a restart finds manual and webhook runs again and records how they en WHERE task_id = ${id} `; } - for (const messageId of Object.values(messages)) { + for (const messageId of [...Object.values(followed), matched]) { yield* sql` INSERT INTO orchestration_v2_projection_messages (message_id, thread_id, run_id, role, streaming, created_at, updated_at, payload_json) VALUES (${messageId}, ${threadId}, ${`run:${messageId}`}, 'user', 0, ${startedAt}, ${startedAt}, '{}') `; } + // Already recorded as the clean success it was, so there is nothing to follow. + yield* sql` + INSERT INTO orchestration_v2_projection_runs + (run_id, thread_id, ordinal, provider, status, requested_at, payload_json) + VALUES (${`run:${matched}`}, ${threadId}, 1, 'codex', 'completed', ${startedAt}, '{}') + `; const layer = followerLayer( Layer.mock(ThreadManagementService.ThreadManagementService)({ getThreadShell: () => Effect.succeed(threadShell(ProjectId.make("project:test"))), getProjectThreadRecords: () => Effect.succeed({ - runs: Object.values(messages).map((userMessageId) => ({ + runs: Object.values(followed).map((userMessageId) => ({ id: `run:${userMessageId}`, status: "failed", rootNodeId: "node:root", userMessageId, })), - messages: Object.values(messages).map((id) => ({ id, runId: `run:${id}` })), - turnItems: Object.values(messages).map((id) => ({ + messages: Object.values(followed).map((id) => ({ id, runId: `run:${id}` })), + turnItems: Object.values(followed).map((id) => ({ type: "error", status: "failed", runId: `run:${id}`, @@ -743,31 +753,40 @@ it.effect("a restart finds manual and webhook runs again and records how they en })), } as never), waitForThread: (input) => - Effect.succeed({ - threadId: input.threadId, - run: { id: input.runId, status: "failed", rootNodeId: "node:root" }, - timedOut: false, - } as never), + Deferred.await(release).pipe( + Effect.as({ + threadId: input.threadId, + run: { id: input.runId, status: "failed", rootNodeId: "node:root" }, + timedOut: false, + } as never), + ), }), ); yield* Effect.gen(function* () { const service = yield* ScheduledTaskService.ScheduledTaskService; + const pending = (yield* service.list()).tasks + .filter((task) => task.runInProgressSince !== undefined) + .map((task) => task.id) + .toSorted(); + assert.deepEqual(pending, ["task:manual", "task:scheduled", "task:webhook"]); const settled = yield* service.subscribeList().pipe( Stream.map(({ tasks }) => tasks), - Stream.filter((tasks) => - tasks.every((task) => task.id === "task:lost" || task.lastRunStatus === "failed"), - ), + Stream.filter((tasks) => tasks.every((task) => task.runInProgressSince === undefined)), Stream.runHead, + Effect.forkChild, ); + yield* Deferred.succeed(release, undefined); const recorded = Object.fromEntries( - Option.getOrThrow(settled).map((task) => [ + Option.getOrThrow(yield* Fiber.join(settled)).map((task) => [ task.id, [task.lastRunStatus, task.lastRunError], ]), ); assert.deepEqual(recorded, { + "task:scheduled": ["failed", PREPARATION_FAILURE], "task:manual": ["failed", PREPARATION_FAILURE], "task:webhook": ["failed", PREPARATION_FAILURE], + "task:matched": ["succeeded", null], // No message to follow: left as it was. "task:lost": ["succeeded", null], }); diff --git a/apps/server/src/scheduledTasks/ScheduledTaskService.ts b/apps/server/src/scheduledTasks/ScheduledTaskService.ts index 2cfed6f3f5c60..206cf4c3ce113 100644 --- a/apps/server/src/scheduledTasks/ScheduledTaskService.ts +++ b/apps/server/src/scheduledTasks/ScheduledTaskService.ts @@ -2,7 +2,6 @@ import { CommandId, MessageId, type OrchestrationV2Run, - OrchestrationV2RunStatus, type RunId, PROVIDER_SEND_TURN_MAX_INPUT_CHARS, type ProjectId, @@ -18,7 +17,7 @@ import { type ScheduledTaskListWebhookDeliveriesInput, type ScheduledTaskListWebhookDeliveriesResult, type ScheduledTaskRotateWebhookTokenInput, - type ScheduledTaskRunStatus, + ScheduledTaskRunStatus, type ScheduledTaskWebhookDeliveryOutcome, type ScheduledTaskWebhookFilter, type ScheduledTaskListResult, @@ -243,6 +242,7 @@ interface LastRunRow { readonly task_id: string; readonly last_run_at: string; readonly last_run_status: string; + readonly last_run_error: string | null; readonly last_run_webhook_delivery_id: string | null; } @@ -332,7 +332,7 @@ const THREAD_DELETED_OUTCOME: RunOutcome = { status: "failed", error: "The run's thread was deleted before the run finished.", }; -const isRunStatus = Schema.is(OrchestrationV2RunStatus); +const isStoredRunStatus = Schema.is(ScheduledTaskRunStatus); const localNow = DateTime.withCurrentZoneLocal(DateTime.nowInCurrentZone); function nextRunAt( @@ -785,24 +785,41 @@ export const layer = Layer.effect( ), ); + // Returns the run state it replaces, read in the same transaction. const markRunning = (id: ScheduledTaskId, startedAtIso: string, webhookDeliveryId?: string) => - sql<{ task_id: string }>` - UPDATE scheduled_tasks - SET updated_at = ${startedAtIso}, - last_run_at = ${startedAtIso}, - last_run_webhook_delivery_id = ${webhookDeliveryId ?? null}, - last_run_status = 'running', - last_run_error = NULL - WHERE task_id = ${id} - RETURNING task_id - `.pipe( - Effect.mapError((cause) => - taskError("Could not mark schedule task as running.", { taskId: id, cause }), - ), - Effect.flatMap((rows) => - rows.length > 0 ? Effect.void : taskError("Schedule task not found.", { taskId: id }), - ), - ); + sql + .withTransaction( + Effect.gen(function* () { + const [previous] = yield* sql<{ + last_run_status: string; + last_run_error: string | null; + }>`SELECT last_run_status, last_run_error FROM scheduled_tasks WHERE task_id = ${id}`; + if (previous === undefined) { + return yield* taskError("Schedule task not found.", { taskId: id }); + } + yield* sql` + UPDATE scheduled_tasks + SET updated_at = ${startedAtIso}, + last_run_at = ${startedAtIso}, + last_run_webhook_delivery_id = ${webhookDeliveryId ?? null}, + last_run_status = 'running', + last_run_error = NULL + WHERE task_id = ${id} + `; + const status = + isStoredRunStatus(previous.last_run_status) && previous.last_run_status !== "running" + ? previous.last_run_status + : ("never" as const); + return { status, error: previous.last_run_error }; + }), + ) + .pipe( + Effect.mapError((cause) => + isScheduledTaskError(cause) + ? cause + : taskError("Could not mark schedule task as running.", { taskId: id, cause }), + ), + ); const markCompleted = (input: { readonly id: ScheduledTaskId; @@ -1038,13 +1055,13 @@ export const layer = Layer.effect( LIMIT 1 `.pipe(NodeSqliteClient.readOnly); if (found === undefined) return; + // A failure's error text needs the run's turn items, so only a clean success is skipped. if ( - isRunStatus(found.run_status) && - ThreadManagementService.isTerminalRunStatus(found.run_status) - ) { - const recorded = found.run_status === "completed" ? "succeeded" : "failed"; - if (row.last_run_status === recorded) return; - } + found.run_status === "completed" && + row.last_run_status === "succeeded" && + row.last_run_error === null + ) + return; const taskId = ScheduledTaskId.make(row.task_id); yield* setPendingRun(taskId, row.last_run_at); yield* followRunOutcome( @@ -1153,21 +1170,27 @@ export const layer = Layer.effect( if (reason !== null) return yield* new WebhookDeliverySkipped({ reason }); } - if (webhook === undefined) yield* markRunning(active.id, startedAtIso); - else - yield* sql - .withTransaction( - Effect.gen(function* () { - yield* markRunning(active.id, startedAtIso, webhook.deliveryId); - // Recovery must distinguish this delivery from other windows waiting on the same task. - yield* sql`UPDATE scheduled_task_webhook_pending SET state = 'running' WHERE delivery_id = ${webhook.deliveryId}`; - }), - ) - .pipe( - Effect.mapError((cause) => - taskError("Could not start webhook run.", { taskId: task.id, cause }), - ), - ); + const previousRun = + webhook === undefined + ? yield* markRunning(active.id, startedAtIso) + : yield* sql + .withTransaction( + Effect.gen(function* () { + const previous = yield* markRunning( + active.id, + startedAtIso, + webhook.deliveryId, + ); + // Recovery must distinguish this delivery from other windows waiting on the same task. + yield* sql`UPDATE scheduled_task_webhook_pending SET state = 'running' WHERE delivery_id = ${webhook.deliveryId}`; + return previous; + }), + ) + .pipe( + Effect.mapError((cause) => + taskError("Could not start webhook run.", { taskId: task.id, cause }), + ), + ); yield* notifyChanged; @@ -1262,12 +1285,8 @@ export const layer = Layer.effect( if (webhook !== undefined && runSucceeded) yield* Ref.set(webhook.dispatched, true); const completedAt = yield* localNow; // An accepted run keeps the previous run's status until its own outcome is known. - const lastRunStatus = !runSucceeded - ? ("failed" as const) - : active.lastRunStatus === "running" - ? ("never" as const) - : active.lastRunStatus; - const lastRunError = runSucceeded ? active.lastRunError : errorMessage(result.cause); + const lastRunStatus = runSucceeded ? previousRun.status : ("failed" as const); + const lastRunError = runSucceeded ? previousRun.error : errorMessage(result.cause); const threadId = runSucceeded ? result.value : null; // One transaction, so an edit cannot land between reading the schedule and // writing the next run computed from it. The origin is read first: it reads @@ -1395,7 +1414,14 @@ 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}`} + ${ + runError === undefined + ? sql`` + : // A new last_run_at makes the miss the latest run, so no earlier run's follower, + // now or after a restart, can record its outcome over it. + sql`, last_run_status = 'failed', last_run_error = ${runError}, + last_run_at = ${iso(now)}, last_run_webhook_delivery_id = NULL` + } WHERE task_id = ${task.id} AND next_run_at IS ${task.nextRunAt} AND last_run_at IS ${task.lastRunAt} @@ -1406,7 +1432,6 @@ export const layer = Layer.effect( ), ); if (updated.length === 0) return; - // The recorded miss stands over the outcome of a run still pending from before it. if (runError !== undefined) yield* clearPendingRun(task.id); yield* notifyChanged; }); @@ -1464,7 +1489,7 @@ export const layer = Layer.effect( // Before the running-row recovery below, whose rows never dispatched a message to follow. yield* sql` - SELECT task_id, last_run_at, last_run_status, last_run_webhook_delivery_id + SELECT task_id, last_run_at, last_run_status, last_run_error, last_run_webhook_delivery_id FROM scheduled_tasks WHERE last_run_at IS NOT NULL AND last_run_status <> 'running' `.pipe( diff --git a/apps/web/src/components/chat/ThreadAutomationsPanel.tsx b/apps/web/src/components/chat/ThreadAutomationsPanel.tsx index 6b3ac2f5d67aa..f1f610a29eb89 100644 --- a/apps/web/src/components/chat/ThreadAutomationsPanel.tsx +++ b/apps/web/src/components/chat/ThreadAutomationsPanel.tsx @@ -224,7 +224,9 @@ export function ThreadAutomationsPanel(props: { part="icon" aria-label={`Run ${task.title} now`} disabled={ - !canOperate || busyTaskId !== null || task.lastRunStatus === "running" + !canOperate || + busyTaskId !== null || + scheduledTaskRunDisplayStatus(task) === "running" } onClick={() => void runNow(task)} > diff --git a/apps/web/src/components/settings/ScheduledTasksSettings.tsx b/apps/web/src/components/settings/ScheduledTasksSettings.tsx index 18791f4c8a6ab..f37bdc1d91c3d 100644 --- a/apps/web/src/components/settings/ScheduledTasksSettings.tsx +++ b/apps/web/src/components/settings/ScheduledTasksSettings.tsx @@ -800,7 +800,9 @@ function ScheduledTaskRow({ {task.threadId !== null && scheduledTaskThreadMode(task) !== "new_thread" ? ( ) : null} - {task.lastRunError ? {task.lastRunError} : null} + {task.lastRunError && task.runInProgressSince === undefined ? ( + {task.lastRunError} + ) : null} } control={