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..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 }), @@ -2169,9 +2170,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/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 faa8d8b8b3cbf..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, "succeeded"); + 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 bf37cdb29d1ea..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, "succeeded"); + 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, "succeeded"); + 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, "succeeded"); + 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 92167ef83c5ca..26786628e55b7 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,128 @@ 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))); + const missedError = 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 }); + 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)), +); + it.effect( "a run made while the scheduler was busy is not overwritten by its stale missed slot", () => @@ -549,8 +675,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: "succeeded", + 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 53ea639975bd2..342688b025959 100644 --- a/apps/server/src/scheduledTasks/ScheduledTaskService.test.ts +++ b/apps/server/src/scheduledTasks/ScheduledTaskService.test.ts @@ -2,10 +2,12 @@ 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, ProviderInstanceId, + type ScheduledTask, ScheduledTaskError, ScheduledTaskId, type ScheduledTaskUpsertInput, @@ -15,14 +17,17 @@ 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"; 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"; @@ -419,58 +424,431 @@ 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."; +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([ - { 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, + }, + { + name: "is steered into the active run, which completes", + run: "cancelled", + status: "succeeded", + error: null, + joins: true, + }, + { + name: "loses its soft-deleted thread", + run: "completed", + status: "failed", + 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* () { 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.status, rootNodeId: "node:root" }; + const run = { id: "run:2", status: outcome.run, rootNodeId: "node:root" }; + 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", + }, + }, + ], + }) 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; + }), + }), + 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; + const { task } = yield* service.upsert({ + id: ScheduledTaskId.make("task:quota"), + title: "Quota bookkeeping", + prompt: "Update the notes", + enabled: true, + schedule: { type: "interval", everyMs: 60_000 }, + projectId, + threadId: unbound ? null : threadId, + workspaceStrategy: { type: "root" }, + modelSelection: { instanceId: ProviderInstanceId.make("codex"), model: "gpt-5.4" }, + runtimeMode: "full-access", + interactionMode: "default", + }); + const dispatched = yield* service.runNow({ id: task.id }); + // 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?.runInProgressSince === undefined), + Stream.runHead, + Effect.forkChild, + ); + yield* Deferred.succeed(ended, undefined); + const recorded = Option.getOrUndefined(yield* Fiber.join(settled)); + 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 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 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 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", + }; + 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 + 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(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(followed).map((userMessageId) => ({ + id: `run:${userMessageId}`, + status: "failed", + rootNodeId: "node:root", + userMessageId, + })), + messages: Object.values(followed).map((id) => ({ id, runId: `run:${id}` })), + turnItems: Object.values(followed).map((id) => ({ + type: "error", + status: "failed", + runId: `run:${id}`, + nodeId: "node:root", + failure: { message: PREPARATION_FAILURE }, + })), + } as never), + waitForThread: (input) => + 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.runInProgressSince === undefined)), + Stream.runHead, + Effect.forkChild, + ); + yield* Deferred.succeed(release, undefined); + const recorded = Object.fromEntries( + 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], + }); + }).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: "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 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) + 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)({}), + Layer.mock(ThreadLaunchService.ThreadLaunchService)({ + launch: () => + Ref.set(dispatched, "launched").pipe( + Effect.as({ threadId: ThreadId.make("thread:launched") } as never), + ), + }), Layer.mock(ThreadManagementService.ThreadManagementService)({ - sendToThread: (input) => Ref.set(sent, input.messageId).pipe(Effect.as({} as never)), + sendToThread: () => Ref.set(dispatched, "sent").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: "run:2" }], - turnItems: [ - { - type: "error", - status: "failed", - runId: "run:2", - nodeId: "node:root", - failure: { message: "Weekly usage limit reached" }, - }, - ], - }) as never, - ), - waitForThread: (input) => - Deferred.await(ended).pipe( - Effect.as({ - threadId: input.threadId, - run: { ...run, id: input.runId, userMessageId: "" }, - timedOut: false, - } as never), - ), + Effect.succeed({ runs: [], messages: [], turnItems: [] } as never), }), Layer.mock(SecretRequests.SecretRequests)({}), Layer.mock(ManagedProjectFolders.ManagedProjectFolders)({ namedProjectsRoot: "/projects", - isScratchProject: () => Effect.succeed(false), + isScratchProject: () => Effect.succeed("scratch" in row), }), NodeCrypto.layer, Scheduler.layer, @@ -479,32 +857,42 @@ it.effect.each([ ); yield* Effect.gen(function* () { const service = yield* ScheduledTaskService.ScheduledTaskService; - const { task } = yield* service.upsert({ - id: ScheduledTaskId.make("task:quota"), - title: "Quota bookkeeping", + const input: ScheduledTaskUpsertInput = { + id: ScheduledTaskId.make("task:repo"), + title: "Nightly notes", prompt: "Update the notes", enabled: true, schedule: { type: "interval", everyMs: 60_000 }, projectId, - threadId: ThreadId.make("thread:quota"), + 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", - }); - yield* service.runNow({ id: task.id }); - const failed = yield* service.subscribeList().pipe( - Stream.map(({ tasks }) => tasks.find((candidate) => candidate.id === task.id)), - Stream.filter((current) => current?.lastRunStatus === "failed"), - Stream.runHead, - Effect.forkChild, - ); - yield* Deferred.succeed(ended, undefined); - const recorded = Option.getOrUndefined(yield* Fiber.join(failed)); - assert.equal(recorded?.lastRunError, outcome.error); - assert.equal(recorded?.runCount, 1); + 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.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.runInProgressSince, result.task.lastRunAt); + } }).pipe(Effect.provide(layer)); - }).pipe(Effect.provide(SqlitePersistence.layerMemory)), + }).pipe(Effect.provide(Layer.merge(SqlitePersistence.layerMemory, NodeServices.layer))), ); it.effect("a bound run follows its thread when promotion moves it mid-send", () => diff --git a/apps/server/src/scheduledTasks/ScheduledTaskService.ts b/apps/server/src/scheduledTasks/ScheduledTaskService.ts index 74d519615bcc2..206cf4c3ce113 100644 --- a/apps/server/src/scheduledTasks/ScheduledTaskService.ts +++ b/apps/server/src/scheduledTasks/ScheduledTaskService.ts @@ -1,6 +1,8 @@ import { CommandId, MessageId, + type OrchestrationV2Run, + type RunId, PROVIDER_SEND_TURN_MAX_INPUT_CHARS, type ProjectId, ScheduledTask, @@ -15,6 +17,7 @@ import { type ScheduledTaskListWebhookDeliveriesInput, type ScheduledTaskListWebhookDeliveriesResult, type ScheduledTaskRotateWebhookTokenInput, + ScheduledTaskRunStatus, type ScheduledTaskWebhookDeliveryOutcome, type ScheduledTaskWebhookFilter, type ScheduledTaskListResult, @@ -59,6 +62,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 { findGitEntry } from "../mcp/gitEntryProbe.ts"; import { isMissedFixedTimeRun, isSameSchedule, nextScheduledRunAt } from "./Schedule.ts"; import { redactHeaders, @@ -234,6 +238,14 @@ interface ScheduledTaskRow { readonly webhook_secret: string | null; } +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; +} + interface WebhookDeliveryRow { readonly delivery_id: string; readonly task_id: string; @@ -315,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 isStoredRunStatus = Schema.is(ScheduledTaskRunStatus); const localNow = DateTime.withCurrentZoneLocal(DateTime.nowInCurrentZone); function nextRunAt( @@ -480,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(), ); @@ -545,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 }, ); }); @@ -590,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) => @@ -754,30 +785,47 @@ 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; readonly completedAtIso: string; readonly nextRunAtIso: string | null; - readonly status: "succeeded" | "failed"; + readonly status: Exclude; readonly error: string | null; readonly startedAtIso: string; }) => @@ -880,52 +928,177 @@ 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 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, + outcome: RunOutcome, + ) { + if ((yield* Ref.get(pendingRuns)).get(taskId) !== startedAtIso) return; + yield* sql` + UPDATE scheduled_tasks + 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' + `; + }); + + // 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 || shell.deletedAt != null) return THREAD_DELETED_OUTCOME; + const thread = { projectId: shell.projectId, threadId }; + const { runs, messages } = yield* threadManagement.getProjectThreadRecords(thread, [ + "runs", + "messages", + ]); + 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; 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"], + ); + 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 + ) { + const error = + latestRootProviderFailure(run, turnItems)?.message ?? + UNSUCCESSFUL_RUN_ERRORS[run.status]; + return { status: "failed", error } as const; + } + runId = joined; + } + return { status: "failed", error: UNSUCCESSFUL_RUN_ERRORS.cancelled } as const; + }, + Effect.catchTags({ + ThreadManagementThreadNotFoundError: () => Effect.succeed(THREAD_DELETED_OUTCOME), + ThreadManagementRunNotFoundError: () => + Effect.succeed({ + status: "failed", + error: "The run was removed before it finished.", + } as const), + }), + ); + + // 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 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. + 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 [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; + // A failure's error text needs the run's turn items, so only a clean success is skipped. if ( - run.status === "cancelled" && - messages.find((message) => message.id === messageId)?.runId !== run.id + found.run_status === "completed" && + row.last_run_status === "succeeded" && + row.last_run_error === null ) return; - const error = - latestRootProviderFailure(run, turnItems)?.message ?? UNSUCCESSFUL_RUN_ERRORS[run.status]; - const updated = yield* sql` - UPDATE scheduled_tasks - SET last_run_status = 'failed', last_run_error = ${error} - WHERE task_id = ${taskId} AND last_run_at = ${startedAtIso} AND last_run_status = 'succeeded' - RETURNING task_id - `; - if (updated.length > 0) yield* notifyChanged; + const taskId = ScheduledTaskId.make(row.task_id); + yield* setPendingRun(taskId, row.last_run_at); + 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) return; + const gitEntry = yield* findGitEntry(project.workspace_root); + if (gitEntry === null) return; + return yield* taskError( + 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 }, + ); + }, + ); + const runTask = Effect.fn("ScheduledTaskService.runTask")(function* ( task: ScheduledTask, trigger: "scheduled" | "manual" | "webhook", @@ -958,11 +1131,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 ( @@ -1001,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; @@ -1069,30 +1244,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,8 +1284,9 @@ 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 lastRunError = runSucceeded ? null : errorMessage(result.cause); + // An accepted run keeps the previous run's status until its own outcome is known. + 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 @@ -1171,20 +1351,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) { - yield* recordRunFailure(active.id, startedAtIso, threadId, messageId).pipe( - Effect.catchCause((cause) => - Effect.logWarning("Could not record scheduled run outcome", { - taskId: active.id, - cause, - }), - ), + 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), @@ -1234,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} @@ -1244,7 +1431,9 @@ 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; + if (runError !== undefined) yield* clearPendingRun(task.id); + yield* notifyChanged; }); const runDueTasks = Effect.fn("ScheduledTaskService.runDueTasks")(function* () { @@ -1298,6 +1487,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_error, 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(); @@ -1561,7 +1777,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 e06ed817ac4f4..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 = 'succeeded' 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, "succeeded"); + // 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 = 'succeeded' 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 = 'succeeded' 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 1bec327103211..f1f610a29eb89 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 { @@ -170,7 +171,7 @@ export function ThreadAutomationsPanel(props: { @@ -223,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 848b88130f48a..f37bdc1d91c3d 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"; @@ -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,13 +794,15 @@ function ScheduledTaskRow({ "Not scheduled" )} - {task.lastRunStatus !== "never" ? ( - {task.lastRunStatus} + {runStatus !== "never" ? ( + {runStatus} ) : null} {task.threadId !== null && scheduledTaskThreadMode(task) !== "new_thread" ? ( ) : null} - {task.lastRunError ? {task.lastRunError} : null} + {task.lastRunError && task.runInProgressSince === undefined ? ( + {task.lastRunError} + ) : null} } control={ 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 4297ad94b73e9..5beb4b0672ef5 100644 --- a/packages/contracts/src/scheduledTask.ts +++ b/packages/contracts/src/scheduledTask.ts @@ -263,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;