From 6be5346c79e65423d474a724f0e86bd685fd4607 Mon Sep 17 00:00:00 2001 From: Hasan Date: Thu, 15 Oct 2026 08:17:28 +0400 Subject: [PATCH 1/5] fix(server): scheduled runs that cannot get a worktree are reported as failed A scheduled task that launches a fresh thread per run with a worktree strategy was recorded as "succeeded" as soon as the thread launch was accepted, even when its worktree could never be made, such as in a folder that is not a Git repository. The run then failed inside its thread before the agent started, and the scheduler never said so. - The pre-launch worktree check now also refuses a folder with no `.git` in it or above it. A repository Git refuses to read is still left to the launch. - The scheduler runs that check before launching, so such a run is recorded as failed with the reason, and no dead thread is created. - schedule_task and update_scheduled_task take an optional workspace (root or worktree), so an agent can choose to run in the project folder; an explicit worktree that cannot be made is refused up front. The task summary now shows the workspace, and an update keeps it unless the binding or the workspace changes. Co-Authored-By: Claude Opus 5.5 --- .../src/mcp/OrchestratorMcpService.test.ts | 98 +++++++++++++++++++ apps/server/src/mcp/OrchestratorMcpService.ts | 54 +++++++--- .../src/mcp/toolkits/orchestrator/tools.ts | 4 +- .../src/mcp/toolkits/project/handlers.ts | 13 ++- .../ThreadLaunchService.test.ts | 53 ++++++++-- .../orchestration-v2/ThreadLaunchService.ts | 71 +++++++++++--- .../ScheduledTaskService.test.ts | 48 ++++++++- .../scheduledTasks/ScheduledTaskService.ts | 42 ++++---- .../ScheduledTaskService.webhook.test.ts | 1 + packages/contracts/src/orchestratorMcp.ts | 10 ++ 10 files changed, 328 insertions(+), 66 deletions(-) diff --git a/apps/server/src/mcp/OrchestratorMcpService.test.ts b/apps/server/src/mcp/OrchestratorMcpService.test.ts index 64f2d86eda94..b00940aefb4b 100644 --- a/apps/server/src/mcp/OrchestratorMcpService.test.ts +++ b/apps/server/src/mcp/OrchestratorMcpService.test.ts @@ -18,6 +18,7 @@ import { import * as DateTime from "effect/DateTime"; import * as Effect from "effect/Effect"; import * as Layer from "effect/Layer"; +import * as Option from "effect/Option"; import * as Ref from "effect/Ref"; import { @@ -2175,6 +2176,103 @@ describe("OrchestratorMcpService provider resolution", () => { assert.equal(updated.webhookUrl, undefined); }), ); + + // Saves echo the task back; the project folder is a Git repository or not. + const workspaceService = (repo: boolean, tasks: ReadonlyArray) => + OrchestratorMcpService.layer.pipe( + Layer.provide( + Layer.mock(ThreadLaunchService.ThreadLaunchService)({ + checkWorktreeBase: () => + repo + ? Effect.void + : Effect.fail( + new ThreadLaunchService.ThreadLaunchNotRepositoryError({ + projectId, + projectTitle: "No project", + workspaceRoot: "/scratch", + }), + ), + }), + ), + Layer.provide(Layer.mock(GitVcsDriver.GitVcsDriver)({})), + Layer.provide( + Layer.mergeAll( + NodeServices.layer, + Layer.mock(ThreadManagementService.ThreadManagementService)({}), + ServerSettings.layerTest(), + Layer.mock(ProviderRegistry.ProviderRegistry)({ getProviders: Effect.succeed([]) }), + Layer.mock(ProviderAdapterRegistry.ProviderAdapterRegistryV2)({ + list: () => Effect.succeed([]), + }), + Layer.mock(ProjectService.ProjectService)({ + getById: (id) => + Effect.succeed( + Option.some({ id, defaultModelSelection: task({}).modelSelection } as never), + ), + }), + Layer.mock(SecretRequests.SecretRequests)({}), + Layer.mock(ScheduledTaskService.ScheduledTaskService)({ + list: () => Effect.succeed({ tasks }), + upsert: (input) => + Effect.succeed({ task: task({ workspaceStrategy: input.workspaceStrategy }) }), + }), + ), + ), + ); + + it.effect.each([ + { + name: "a worktree in a Git repository", + repo: true, + workspace: "worktree", + want: "worktree", + }, + { name: "root when asked for", repo: true, workspace: "root", want: "root" }, + { + name: "a worktree in a plain folder is refused", + repo: false, + workspace: "worktree", + want: "invalid_request", + }, + ] as const)("an unbound task over MCP: $name", (testCase) => + Effect.gen(function* () { + const outcome = yield* OrchestratorMcpService.OrchestratorMcpService.pipe( + Effect.flatMap((mcp) => + mcp.scheduleTask(supervisedClient, { + projectId, + prompt: "Check the inbox.", + schedule: { type: "interval", everyMs: 3_600_000 }, + workspace: testCase.workspace, + }), + ), + Effect.map((summary) => summary.workspace), + Effect.catch((error) => Effect.succeed(error.code)), + Effect.provide(workspaceService(testCase.repo, [])), + ); + assert.equal(outcome, testCase.want); + }), + ); + + it.effect("an update keeps the workspace unless the binding or the workspace changes", () => + Effect.gen(function* () { + // An unbound task that chose root in a Git repository. + const existing = task({ workspaceStrategy: { type: "root" } }); + const mcp = yield* OrchestratorMcpService.OrchestratorMcpService.pipe( + Effect.provide(workspaceService(true, [existing])), + ); + const repeated = yield* mcp.updateScheduledTask(supervisedClient, { + scheduledTaskId: existing.id, + prompt: "Check the inbox again.", + bindToCurrentThread: false, + }); + assert.equal(repeated.workspace, "root"); + const switched = yield* mcp.updateScheduledTask(supervisedClient, { + scheduledTaskId: existing.id, + workspace: "worktree", + }); + assert.equal(switched.workspace, "worktree"); + }), + ); }); describe("resolveRuntimeMode", () => { diff --git a/apps/server/src/mcp/OrchestratorMcpService.ts b/apps/server/src/mcp/OrchestratorMcpService.ts index 5c0a93cb1c0f..0f194936453c 100644 --- a/apps/server/src/mcp/OrchestratorMcpService.ts +++ b/apps/server/src/mcp/OrchestratorMcpService.ts @@ -229,19 +229,6 @@ function errorMessage(error: unknown): string { return error instanceof Error ? error.message : String(error); } -/** - * Workspace strategy for a scheduled task created/updated over MCP: bound runs - * post into the existing thread (the strategy is unused, keep root); unbound - * runs launch a fresh worktree per run. - */ -function scheduledTaskWorkspaceStrategy( - boundToThread: boolean, -): ScheduledTask["workspaceStrategy"] { - return boundToThread - ? { type: "root" } - : { type: "worktree", baseRef: "main", startFromOrigin: true }; -} - /** * A scheduled task as an agent sees it. `mayRun` says whether the caller may * run it: a webhook's URL carries the secret that starts the task's runs, so @@ -255,6 +242,7 @@ function scheduledTaskSummary(task: ScheduledTask, mayRun: boolean): Orchestrato enabled: task.enabled, projectId: task.projectId, boundThreadId: task.threadId, + workspace: task.workspaceStrategy.type, schedule: task.schedule, nextRunAt: task.nextRunAt, lastRunStatus: task.lastRunStatus, @@ -1635,6 +1623,31 @@ const make = Effect.gen(function* () { return task; }); + /** + * Workspace strategy for a scheduled task created/updated over MCP. Bound + * runs post into the existing thread, so the strategy is unused and stays + * root. Unbound runs get a fresh worktree unless root is asked for; an + * explicit worktree is refused up front when one cannot be made. + */ + const scheduledTaskWorkspaceStrategy = ( + projectId: ScheduledTask["projectId"], + boundToThread: boolean, + workspace: "root" | "worktree" | undefined, + ): Effect.Effect => { + if (boundToThread || workspace === "root") return Effect.succeed({ type: "root" }); + const worktree = { type: "worktree", baseRef: "main", startFromOrigin: true } as const; + if (workspace === undefined) return Effect.succeed(worktree); + return threadLaunch.checkWorktreeBase({ projectId, workspaceStrategy: worktree }).pipe( + Effect.as(worktree), + Effect.mapError((error) => + failure( + "invalid_request", + `${error.message} Pass workspace:"root" to run in the project folder.`, + ), + ), + ); + }; + return OrchestratorMcpService.of({ scheduleTask: (scope, input) => Effect.gen(function* () { @@ -1669,7 +1682,11 @@ const make = Effect.gen(function* () { schedule: input.schedule, projectId, threadId: bindToCurrentThread && parent !== undefined ? parent.thread.id : null, - workspaceStrategy: scheduledTaskWorkspaceStrategy(bindToCurrentThread), + workspaceStrategy: yield* scheduledTaskWorkspaceStrategy( + projectId, + bindToCurrentThread, + input.workspace, + ), modelSelection, runtimeMode: limits.runtimeMode, interactionMode: limits.interactionMode, @@ -1739,10 +1756,15 @@ const make = Effect.gen(function* () { // Rebinding changes where runs execute, so the workspace strategy must // follow: unbinding a root-strategy task would otherwise run loose // prompts in the shared project checkout. + const rebound = (threadId === null) !== (existing.threadId === null); const workspaceStrategy = - input.bindToCurrentThread === undefined + !rebound && input.workspace === undefined ? existing.workspaceStrategy - : scheduledTaskWorkspaceStrategy(input.bindToCurrentThread); + : yield* scheduledTaskWorkspaceStrategy( + existing.projectId, + threadId !== null, + input.workspace, + ); const upsertInput: ScheduledTaskUpsertInput = { id: existing.id, title: input.title ?? existing.title, diff --git a/apps/server/src/mcp/toolkits/orchestrator/tools.ts b/apps/server/src/mcp/toolkits/orchestrator/tools.ts index 36c4d78723c0..89fda79a72d0 100644 --- a/apps/server/src/mcp/toolkits/orchestrator/tools.ts +++ b/apps/server/src/mcp/toolkits/orchestrator/tools.ts @@ -116,7 +116,7 @@ const WorkflowStopTool = Tool.make("t3_workflow_stop", { export const ScheduleTaskTool = Tool.make("schedule_task", { description: - "Create persistent work in the app scheduler that runs even when no turn is active. Pass schedule as a STRUCTURED OBJECT, never JSON text. Timers: {type:'interval', everyMs:3600000} is hourly; {type:'fixed_time', timeOfDay:'09:00', weekdays:[1,2,3,4,5]} is weekday mornings; report the returned nextRunAt. Webhooks: {type:'webhook'} runs once per request to a generated URL. The run sees the request ONLY through prompt placeholders: {{body.path}} (e.g. {{body.action}}, {{body.release.tag_name}}), {{headers.name}}, {{query.name}}, {{body}}, or {{request}} (method, headers with credentials redacted, and body). For a sender that signs requests, first call request_secret so the user enters the secret privately (never ask for it in chat or invent one), then set signature with the returned secretRef, e.g. GitHub: {type:'webhook', signature:{header:'x-hub-signature-256', encoding:'hex', prefix:'sha256=', secretRef}}. The result's webhookUrl is the public URL to give the user; if it is absent, this environment has no T3 Connect managed tunnel, so tell the user to enable T3 Connect remote access rather than sharing a path. Omit projectId for this thread's project. In this thread's project, runs post into THIS thread by default (bindToCurrentThread=true), which suits an orchestrator that sees every trigger, delegates work, and can dedupe against what is in flight; use false only when the user wants a fresh top-level thread per run. Elsewhere each run launches a fresh thread. Provider, model, and runtime settings inherit from this thread, or from the project default when there is no calling thread.", + "Create persistent work in the app scheduler that runs even when no turn is active. Pass schedule as a STRUCTURED OBJECT, never JSON text. Timers: {type:'interval', everyMs:3600000} is hourly; {type:'fixed_time', timeOfDay:'09:00', weekdays:[1,2,3,4,5]} is weekday mornings; report the returned nextRunAt. Webhooks: {type:'webhook'} runs once per request to a generated URL. The run sees the request ONLY through prompt placeholders: {{body.path}} (e.g. {{body.action}}, {{body.release.tag_name}}), {{headers.name}}, {{query.name}}, {{body}}, or {{request}} (method, headers with credentials redacted, and body). For a sender that signs requests, first call request_secret so the user enters the secret privately (never ask for it in chat or invent one), then set signature with the returned secretRef, e.g. GitHub: {type:'webhook', signature:{header:'x-hub-signature-256', encoding:'hex', prefix:'sha256=', secretRef}}. The result's webhookUrl is the public URL to give the user; if it is absent, this environment has no T3 Connect managed tunnel, so tell the user to enable T3 Connect remote access rather than sharing a path. Omit projectId for this thread's project. In this thread's project, runs post into THIS thread by default (bindToCurrentThread=true), which suits an orchestrator that sees every trigger, delegates work, and can dedupe against what is in flight; use false only when the user wants a fresh top-level thread per run. Elsewhere each run launches a fresh thread in a new Git worktree; pass workspace:'root' to run in the project folder instead. Provider, model, and runtime settings inherit from this thread, or from the project default when there is no calling thread.", parameters: OrchestratorMcpScheduleTaskInput, success: OrchestratorMcpScheduleTaskResult, failure: OrchestratorMcpFailure, @@ -143,7 +143,7 @@ const ListScheduledTasksTool = Tool.make("list_scheduled_tasks", { const UpdateScheduledTaskTool = Tool.make("update_scheduled_task", { description: - "Update an existing scheduled task by scheduledTaskId (from list_scheduled_tasks). Only the provided fields change; omit a field to leave it as-is. Use enabled=false to pause a task without deleting it. Set bindToCurrentThread to move the task between posting into this thread and launching a fresh thread per run.", + "Update an existing scheduled task by scheduledTaskId (from list_scheduled_tasks). Only the provided fields change; omit a field to leave it as-is. Use enabled=false to pause a task without deleting it. Set bindToCurrentThread to move the task between posting into this thread and launching a fresh thread per run, and workspace to choose where a fresh-thread run works.", parameters: OrchestratorMcpUpdateScheduledTaskInput, success: OrchestratorMcpScheduleTaskResult, failure: OrchestratorMcpFailure, diff --git a/apps/server/src/mcp/toolkits/project/handlers.ts b/apps/server/src/mcp/toolkits/project/handlers.ts index c346ad62001f..8ac379dc3003 100644 --- a/apps/server/src/mcp/toolkits/project/handlers.ts +++ b/apps/server/src/mcp/toolkits/project/handlers.ts @@ -116,11 +116,14 @@ export const layer = McpToolAccess.toLayer(ProjectToolkit, { (error) => new OrchestratorMcpFailure({ code: "invalid_request", - message: `${error.message} No thread was created. Pass a branch, tag, or commit that exists there${ - strategy.startFromOrigin === true - ? "; the project has no origin remote to fetch it from." - : ", or set startFromOrigin:true if it exists only on origin." - }`, + message: + error._tag === "ThreadLaunchNotRepositoryError" + ? `${error.message} No thread was created. Use workspaceStrategy {type:"root"} instead.` + : `${error.message} No thread was created. Pass a branch, tag, or commit that exists there${ + strategy.startFromOrigin === true + ? "; the project has no origin remote to fetch it from." + : ", or set startFromOrigin:true if it exists only on origin." + }`, }), ), ); diff --git a/apps/server/src/orchestration-v2/ThreadLaunchService.test.ts b/apps/server/src/orchestration-v2/ThreadLaunchService.test.ts index 26e9b8c8d78d..316bd3880514 100644 --- a/apps/server/src/orchestration-v2/ThreadLaunchService.test.ts +++ b/apps/server/src/orchestration-v2/ThreadLaunchService.test.ts @@ -35,6 +35,7 @@ import * as Effect from "effect/Effect"; import * as Exit from "effect/Exit"; import * as Layer from "effect/Layer"; import * as Option from "effect/Option"; +import * as Path from "effect/Path"; import * as Ref from "effect/Ref"; import * as Stream from "effect/Stream"; import * as Schema from "effect/Schema"; @@ -113,6 +114,8 @@ interface HarnessOptions { readonly serverSettings?: Parameters[0]; readonly providers?: ReadonlyArray; readonly git?: Partial; + /** Paths that exist on disk, as far as the launch service can see. */ + readonly existingPaths?: ReadonlyArray; } function makeHarness(options: HarnessOptions = {}) { @@ -182,6 +185,7 @@ function makeHarness(options: HarnessOptions = {}) { renameBranch, fetchRemote: options.fetchRemote ?? (() => Effect.void), hasCommit: options.hasCommit ?? (() => Effect.succeed(false)), + isRepository: () => Effect.succeed(true), remoteExists: () => Effect.succeed(true), remoteBranchExists: () => Effect.succeed(true), removeWorktree, @@ -211,6 +215,10 @@ function makeHarness(options: HarnessOptions = {}) { layerThreadManagement, layerReceipts, IdAllocator.layer, + FileSystem.layerNoop({ + exists: (path) => Effect.succeed(options.existingPaths?.includes(path) === true), + }), + Path.layer, ), ), ); @@ -2525,12 +2533,6 @@ it.effect.each([ startFromOrigin: true, origin: false, }, - { - name: "is a previous checkout there never was", - baseRef: "-", - startFromOrigin: false, - origin: true, - }, ])("refuses a new worktree whose base ref $name", (testCase) => Effect.gen(function* () { const harness = makeHarness({ @@ -2565,8 +2567,41 @@ it.effect.each([ }), ); +it.effect("refuses a new worktree in a folder that is not a Git repository", () => + Effect.gen(function* () { + const harness = makeHarness({ git: { isRepository: () => Effect.succeed(false) } }); + yield* Effect.gen(function* () { + const launches = yield* ThreadLaunch.ThreadLaunchService; + const error = yield* launches + .checkWorktreeBase({ + projectId, + // A commit search skips the base ref check, but not this one. + workspaceStrategy: { type: "worktree", baseRef: ":/fix", startFromOrigin: true }, + }) + .pipe(Effect.flip); + assert.deepInclude(error, { _tag: "ThreadLaunchNotRepositoryError", projectId }); + }).pipe(Effect.provide(harness.layer)); + }), +); + +it.effect("leaves a repository Git refuses to read to the launch", () => + Effect.gen(function* () { + // Git says "not a repository" for one owned by another user, but its `.git` is there. + const harness = makeHarness({ + git: { isRepository: () => Effect.succeed(false) }, + existingPaths: ["/.git"], + }); + yield* Effect.gen(function* () { + const launches = yield* ThreadLaunch.ThreadLaunchService; + yield* launches.checkWorktreeBase({ + projectId, + workspaceStrategy: { type: "worktree", baseRef: ":/fix", startFromOrigin: true }, + }); + }).pipe(Effect.provide(harness.layer)); + }), +); + it.effect.each([ - { name: "a base ref that resolves", baseRef: "main", commits: ["main", "HEAD"] }, // Provisioning fetches it; the ref may exist nowhere else yet. { name: "a base ref only origin has", @@ -2585,7 +2620,7 @@ it.effect.each([ { name: "a repository with no commits yet", baseRef: "main", commits: [] }, { name: "the previous checkout", baseRef: "-", commits: ["@{-1}", "HEAD"] }, { name: "a commit search", baseRef: ":/fix", commits: ["HEAD"] }, - { name: "a folder that is not a repository", baseRef: "main", commits: null }, + { name: "a Git that cannot answer", baseRef: "main", commits: null }, ])("leaves $name to the launch", (testCase) => Effect.gen(function* () { const { commits } = testCase; @@ -2598,7 +2633,7 @@ it.effect.each([ operation: "GitWorkflowService.hasCommit", command: "git rev-parse", cwd: "/repo", - detail: "Not a Git repository.", + detail: "fatal: bad object", }), ) : Effect.succeed(commits.includes(refName)), diff --git a/apps/server/src/orchestration-v2/ThreadLaunchService.ts b/apps/server/src/orchestration-v2/ThreadLaunchService.ts index a4d949770978..1f096e5fb89d 100644 --- a/apps/server/src/orchestration-v2/ThreadLaunchService.ts +++ b/apps/server/src/orchestration-v2/ThreadLaunchService.ts @@ -23,8 +23,10 @@ import * as Cause from "effect/Cause"; import * as Context from "effect/Context"; 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 Option from "effect/Option"; +import * as Path from "effect/Path"; import * as Ref from "effect/Ref"; import * as Schema from "effect/Schema"; import * as Scope from "effect/Scope"; @@ -163,6 +165,20 @@ export class ThreadLaunchBaseRefError extends Schema.TaggedError()( + "ThreadLaunchNotRepositoryError", + { + projectId: ProjectId, + projectTitle: Schema.String, + workspaceRoot: Schema.String, + }, +) { + override get message(): string { + return `Project "${this.projectTitle}" (${this.workspaceRoot}) is not a Git repository, so it cannot get a worktree.`; + } +} + export class ThreadLaunchService extends Context.Service< ThreadLaunchService, { @@ -173,13 +189,14 @@ export class ThreadLaunchService extends Context.Service< ThreadLaunchError | ThreadManagement.ThreadManagementProviderInstanceDisabledError >; /** - * Fails when a new worktree's base ref cannot give provisioning a commit. - * `launch` accepts such a launch and reports the failure inside the - * thread; a caller that needs it before a thread exists checks first. + * Fails when a new worktree cannot be provisioned: the project folder is + * not a Git repository, or the base ref gives it no commit. `launch` + * accepts such a launch and reports the failure inside the thread; a + * caller that needs it before a thread exists checks first. */ readonly checkWorktreeBase: ( input: Pick, - ) => Effect.Effect; + ) => Effect.Effect; /** Dispatches prepared-run.retry and prepares the run's workspace again. */ readonly retryPreparation: (input: ThreadLaunchPreparationInput) => Effect.Effect< Orchestrator.OrchestratorV2DispatchResult, @@ -217,6 +234,8 @@ const make = Effect.gen(function* () { const ids = yield* IdAllocator.IdAllocatorV2; const threads = yield* ThreadManagement.ThreadManagementService; const managedFolders = yield* ManagedProjectFolders.ManagedProjectFolders; + const fileSystem = yield* FileSystem.FileSystem; + const path = yield* Path.Path; const preparationScope = yield* Scope.make("sequential"); const scheduledLaunches = yield* Ref.make>(new Set()); yield* Effect.addFinalizer(() => Scope.close(preparationScope, Exit.void)); @@ -261,19 +280,41 @@ const make = Effect.gen(function* () { } }); + // Git also answers "not a repository" for one it refuses to read, such as a + // checkout owned by another user. Only a folder with no `.git` in it or + // above it is plainly not one. + const hasGitAbove = Effect.fn("ThreadLaunchService.hasGitAbove")(function* (cwd: string) { + let folder = path.resolve(cwd); + while (true) { + if (yield* fileSystem.exists(path.join(folder, ".git"))) return true; + const parent = path.dirname(folder); + if (parent === folder) return false; + folder = parent; + } + }); + const checkWorktreeBase: ThreadLaunchService["Service"]["checkWorktreeBase"] = Effect.fn( "ThreadLaunchService.checkWorktreeBase", )(function* (input) { const strategy = input.workspaceStrategy; - // `:/text` is a commit search, which cannot be asked for a commit type. - if (strategy.type !== "worktree" || strategy.baseRef.startsWith(":")) return; + if (strategy.type !== "worktree") return; const { baseRef } = strategy; // Anything this cannot establish is left for the launch to report: a - // missing project, a folder that is not a repository, a failing Git. - const unresolved = yield* Effect.gen(function* () { + // missing project, a failing Git, a repository Git refuses to read. + const problem = yield* Effect.gen(function* () { const project = Option.getOrUndefined(yield* projects.getById(input.projectId)); if (project === undefined) return undefined; const cwd = project.workspaceRoot; + if (!(yield* git.isRepository(cwd))) + return (yield* hasGitAbove(cwd)) + ? undefined + : new ThreadLaunchNotRepositoryError({ + projectId: project.id, + projectTitle: project.title, + workspaceRoot: cwd, + }); + // `:/text` is a commit search, which cannot be asked for a commit type. + if (baseRef.startsWith(":")) return undefined; // `git worktree add` reads `-` as the previous checkout. if (yield* git.hasCommit({ cwd, refName: baseRef === "-" ? "@{-1}" : baseRef })) return undefined; @@ -287,15 +328,15 @@ const make = Effect.gen(function* () { // `git worktree add` also starts from a remote-tracking branch of that // name; any ref carrying it is reason enough to let Git decide. if (yield* git.hasRefNamed({ cwd, refName: baseRef })) return undefined; - return (yield* git.hasCommit({ cwd, refName: "HEAD" })) ? project : undefined; - }).pipe(Effect.orElseSucceed(() => undefined)); - if (unresolved !== undefined) - return yield* new ThreadLaunchBaseRefError({ - projectId: unresolved.id, - projectTitle: unresolved.title, - workspaceRoot: unresolved.workspaceRoot, + if (!(yield* git.hasCommit({ cwd, refName: "HEAD" }))) return undefined; + return new ThreadLaunchBaseRefError({ + projectId: project.id, + projectTitle: project.title, + workspaceRoot: cwd, baseRef, }); + }).pipe(Effect.orElseSucceed(() => undefined)); + if (problem !== undefined) return yield* problem; }); const prepareInBackground = Effect.fn("ThreadLaunchService.prepareInBackground")(function* ( diff --git a/apps/server/src/scheduledTasks/ScheduledTaskService.test.ts b/apps/server/src/scheduledTasks/ScheduledTaskService.test.ts index 1c95fecdea79..d13741ba70cb 100644 --- a/apps/server/src/scheduledTasks/ScheduledTaskService.test.ts +++ b/apps/server/src/scheduledTasks/ScheduledTaskService.test.ts @@ -3,7 +3,7 @@ import * as NodeUtil from "node:util"; import * as NodeCrypto from "@effect/platform-node/NodeCrypto"; import { assert, it } from "@effect/vitest"; -import { ScheduledTaskError } from "@t3tools/contracts"; +import { ScheduledTaskError, ScheduledTaskUpsertInput } from "@t3tools/contracts"; import * as DateTime from "effect/DateTime"; import * as Effect from "effect/Effect"; import * as Layer from "effect/Layer"; @@ -22,6 +22,7 @@ import * as SqlitePersistence from "../persistence/Sqlite.ts"; import * as ScheduledTaskService from "./ScheduledTaskService.ts"; const isScheduledTaskError = Schema.is(ScheduledTaskError); +const decodeUpsertInput = Schema.decodeEffect(ScheduledTaskUpsertInput); const insertRow = ( sql: SqlClient.SqlClient, @@ -199,6 +200,7 @@ it.effect( ScheduledTaskService.layer, Layer.mergeAll( Layer.mock(ThreadLaunchService.ThreadLaunchService)({ + checkWorktreeBase: () => Effect.void, launch: () => Ref.updateAndGet(dispatched, (n) => n + 1).pipe( Effect.andThen((n) => @@ -297,6 +299,7 @@ it.effect( ScheduledTaskService.layer, Layer.mergeAll( Layer.mock(ThreadLaunchService.ThreadLaunchService)({ + checkWorktreeBase: () => Effect.void, launch: () => Ref.updateAndGet(dispatched, (n) => n + 1).pipe( // A concurrent writer (the CLI shares the SQLite file) @@ -351,3 +354,46 @@ it.effect( ); }).pipe(Effect.provide(SqlitePersistence.layerMemory)), ); + +it.effect("records a run whose worktree cannot be made as failed instead of launching it", () => + Effect.gen(function* () { + const layerDependencies = Layer.mergeAll( + NodeCrypto.layer, + Scheduler.layer, + // No launch: a run that got past the check would die on the mock. + Layer.mock(ThreadLaunchService.ThreadLaunchService)({ + checkWorktreeBase: ({ projectId }) => + Effect.fail( + new ThreadLaunchService.ThreadLaunchNotRepositoryError({ + projectId, + projectTitle: "No project", + workspaceRoot: "/scratch", + }), + ), + }), + Layer.mock(ThreadManagementService.ThreadManagementService)({}), + Layer.mock(SecretRequests.SecretRequests)({}), + ); + yield* Effect.gen(function* () { + const service = yield* ScheduledTaskService.ScheduledTaskService; + const { task } = yield* service.upsert( + yield* decodeUpsertInput({ + title: "Morning check", + prompt: "Check the inbox.", + enabled: true, + schedule: { type: "interval", everyMs: 60_000 }, + projectId: "project-no-repo", + workspaceStrategy: { type: "worktree", baseRef: "main", startFromOrigin: true }, + modelSelection: { instanceId: "codex", model: "gpt-5.4" }, + runtimeMode: "full-access", + interactionMode: "default", + }), + ); + const { task: ran } = yield* service.runNow({ id: task.id }); + assert.equal(ran.lastRunStatus, "failed"); + assert.include(ran.lastRunError, "is not a Git repository"); + const [listed] = (yield* service.list()).tasks; + assert.equal(listed?.lastRunStatus, "failed"); + }).pipe(Effect.provide(ScheduledTaskService.layer.pipe(Layer.provide(layerDependencies)))); + }).pipe(Effect.provide(SqlitePersistence.layerMemory)), +); diff --git a/apps/server/src/scheduledTasks/ScheduledTaskService.ts b/apps/server/src/scheduledTasks/ScheduledTaskService.ts index 086de6ee9258..85981f682f0a 100644 --- a/apps/server/src/scheduledTasks/ScheduledTaskService.ts +++ b/apps/server/src/scheduledTasks/ScheduledTaskService.ts @@ -791,27 +791,33 @@ export const layer = Layer.effect( // Effect.exit (not Effect.result) so defects and interruptions in the // dispatch are also captured and recorded as a failed run instead of - // aborting before markCompleted. + // aborting before markCompleted. A launch is accepted before its + // worktree exists, so a worktree that cannot be made is checked first; + // otherwise the run would count as succeeded and fail inside its thread. 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, - }), + threadLaunch.checkWorktreeBase(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, + }), + ), + ), ) : yield* Effect.exit( threadManagement.sendToThread({ diff --git a/apps/server/src/scheduledTasks/ScheduledTaskService.webhook.test.ts b/apps/server/src/scheduledTasks/ScheduledTaskService.webhook.test.ts index feb8461fa984..398607202e9b 100644 --- a/apps/server/src/scheduledTasks/ScheduledTaskService.webhook.test.ts +++ b/apps/server/src/scheduledTasks/ScheduledTaskService.webhook.test.ts @@ -80,6 +80,7 @@ const withService = ( NodePlatformCrypto.layer, Scheduler.layer, Layer.mock(ThreadLaunchService.ThreadLaunchService)({ + checkWorktreeBase: () => Effect.void, launch: (input) => Queue.offer(launches, input).pipe( Effect.andThen(options.gate ? Deferred.await(options.gate) : Effect.void), diff --git a/packages/contracts/src/orchestratorMcp.ts b/packages/contracts/src/orchestratorMcp.ts index 80328d279794..b1fb94e7553b 100644 --- a/packages/contracts/src/orchestratorMcp.ts +++ b/packages/contracts/src/orchestratorMcp.ts @@ -533,6 +533,11 @@ export const OrchestratorMcpCapabilitiesResult = Schema.Struct({ }); export type OrchestratorMcpCapabilitiesResult = typeof OrchestratorMcpCapabilitiesResult.Type; +const OrchestratorMcpScheduledTaskWorkspace = Schema.Literals(["root", "worktree"]).annotate({ + description: + "Where each fresh-thread run works: 'worktree' (the default) makes a new Git worktree from origin/main; 'root' runs in the project folder (a No-project task gets a new folder per run). Runs that post into this thread ignore it.", +}); + export const OrchestratorMcpScheduleTaskInput = Schema.Struct({ projectId: OrchestratorMcpProjectTarget, prompt: OrchestratorMcpPrompt.annotate({ @@ -554,6 +559,7 @@ export const OrchestratorMcpScheduleTaskInput = Schema.Struct({ "True (default) posts each run into this thread; false creates a fresh top-level thread per run.", }), ), + workspace: Schema.optional(OrchestratorMcpScheduledTaskWorkspace), clientRequestId: Schema.optional(OrchestratorMcpClientRequestId), }); export type OrchestratorMcpScheduleTaskInput = typeof OrchestratorMcpScheduleTaskInput.Type; @@ -566,6 +572,9 @@ export const OrchestratorMcpScheduledTask = Schema.Struct({ enabled: Schema.Boolean, projectId: ProjectId, boundThreadId: Schema.NullOr(ThreadId), + workspace: Schema.Literals(["root", "worktree", "existing_worktree"]).annotate({ + description: "Where each fresh-thread run works.", + }), schedule: ScheduledTaskSchedule, nextRunAt: Schema.NullOr(IsoDateTime), lastRunStatus: ScheduledTaskRunStatus, @@ -607,6 +616,7 @@ export const OrchestratorMcpUpdateScheduledTaskInput = Schema.Struct({ schedule: Schema.optional(OrchestratorMcpSchedule), enabled: Schema.optional(Schema.Boolean), bindToCurrentThread: Schema.optional(Schema.Boolean), + workspace: Schema.optional(OrchestratorMcpScheduledTaskWorkspace), }); export type OrchestratorMcpUpdateScheduledTaskInput = typeof OrchestratorMcpUpdateScheduledTaskInput.Type; From 3e2975783d7cfd4f712e52957353ec76fc9e6e9f Mon Sep 17 00:00:00 2001 From: Hasan Date: Thu, 15 Oct 2026 09:10:47 +0400 Subject: [PATCH 2/5] fix(server): check the worktree only for scheduled worktree runs Root runs have nothing to check, and skipping the call keeps scheduler fixtures that stub only `launch` working. Co-Authored-By: Claude Opus 5.5 --- apps/server/src/scheduledTasks/ScheduledTaskService.test.ts | 2 -- apps/server/src/scheduledTasks/ScheduledTaskService.ts | 5 ++++- .../src/scheduledTasks/ScheduledTaskService.webhook.test.ts | 1 - 3 files changed, 4 insertions(+), 4 deletions(-) diff --git a/apps/server/src/scheduledTasks/ScheduledTaskService.test.ts b/apps/server/src/scheduledTasks/ScheduledTaskService.test.ts index d13741ba70cb..b9a45dcfdbd6 100644 --- a/apps/server/src/scheduledTasks/ScheduledTaskService.test.ts +++ b/apps/server/src/scheduledTasks/ScheduledTaskService.test.ts @@ -200,7 +200,6 @@ it.effect( ScheduledTaskService.layer, Layer.mergeAll( Layer.mock(ThreadLaunchService.ThreadLaunchService)({ - checkWorktreeBase: () => Effect.void, launch: () => Ref.updateAndGet(dispatched, (n) => n + 1).pipe( Effect.andThen((n) => @@ -299,7 +298,6 @@ it.effect( ScheduledTaskService.layer, Layer.mergeAll( Layer.mock(ThreadLaunchService.ThreadLaunchService)({ - checkWorktreeBase: () => Effect.void, launch: () => Ref.updateAndGet(dispatched, (n) => n + 1).pipe( // A concurrent writer (the CLI shares the SQLite file) diff --git a/apps/server/src/scheduledTasks/ScheduledTaskService.ts b/apps/server/src/scheduledTasks/ScheduledTaskService.ts index 85981f682f0a..8fd411ff2d72 100644 --- a/apps/server/src/scheduledTasks/ScheduledTaskService.ts +++ b/apps/server/src/scheduledTasks/ScheduledTaskService.ts @@ -797,7 +797,10 @@ export const layer = Layer.effect( const result = active.threadId === null ? yield* Effect.exit( - threadLaunch.checkWorktreeBase(active).pipe( + (active.workspaceStrategy.type === "worktree" + ? threadLaunch.checkWorktreeBase(active) + : Effect.void + ).pipe( Effect.andThen( threadLaunch.launch({ commandId, diff --git a/apps/server/src/scheduledTasks/ScheduledTaskService.webhook.test.ts b/apps/server/src/scheduledTasks/ScheduledTaskService.webhook.test.ts index 398607202e9b..feb8461fa984 100644 --- a/apps/server/src/scheduledTasks/ScheduledTaskService.webhook.test.ts +++ b/apps/server/src/scheduledTasks/ScheduledTaskService.webhook.test.ts @@ -80,7 +80,6 @@ const withService = ( NodePlatformCrypto.layer, Scheduler.layer, Layer.mock(ThreadLaunchService.ThreadLaunchService)({ - checkWorktreeBase: () => Effect.void, launch: (input) => Queue.offer(launches, input).pipe( Effect.andThen(options.gate ? Deferred.await(options.gate) : Effect.void), From ec0fb71c781b548d56a2f3868362ad5438a2627f Mon Sep 17 00:00:00 2001 From: Hasan Date: Thu, 15 Oct 2026 10:33:38 +0400 Subject: [PATCH 3/5] fix(server): scheduled runs whose workspace preparation fails are reported as failed A launch is accepted before its workspace is prepared, so a run whose worktree fetch or setup script failed still read "succeeded". The scheduler now waits for the launched run to leave preparation and, if it failed, marks that run failed. The write is guarded by the run's start time, so a later run's status is never overwritten. It also covers the hint that t3_thread_launch gives for a folder that is not a Git repository. Co-Authored-By: Claude Opus 5.5 --- .../src/mcp/toolkits/project/handlers.test.ts | 18 ++- .../ScheduledTaskService.test.ts | 66 ++++++++- .../scheduledTasks/ScheduledTaskService.ts | 132 ++++++++++++++---- 3 files changed, 183 insertions(+), 33 deletions(-) diff --git a/apps/server/src/mcp/toolkits/project/handlers.test.ts b/apps/server/src/mcp/toolkits/project/handlers.test.ts index 7b045ae6b189..732055452fb4 100644 --- a/apps/server/src/mcp/toolkits/project/handlers.test.ts +++ b/apps/server/src/mcp/toolkits/project/handlers.test.ts @@ -475,7 +475,15 @@ it.effect("a launch into a new worktree creates nothing when its base ref has no runtimeModeCeiling: "auto", launched, checkWorktreeBase: ({ projectId, workspaceStrategy }) => - workspaceStrategy.type === "worktree" && workspaceStrategy.baseRef === "t3/renamed" + workspaceStrategy.type === "worktree" && workspaceStrategy.baseRef === "plain" + ? Effect.fail( + new ThreadLaunch.ThreadLaunchNotRepositoryError({ + projectId, + projectTitle: "Client target", + workspaceRoot: "/projects/client-target", + }), + ) + : workspaceStrategy.type === "worktree" && workspaceStrategy.baseRef === "t3/renamed" ? Effect.fail( new ThreadLaunch.ThreadLaunchBaseRefError({ projectId, @@ -507,6 +515,14 @@ it.effect("a launch into a new worktree creates nothing when its base ref has no ); expect(launched).toHaveLength(0); + // A folder that is not a repository points the agent at the root instead. + const plain = (yield* launchFrom("plain")).at(-1)?.result; + expect(plain).toMatchObject({ code: "invalid_request" }); + expect((plain as { message: string }).message).toContain( + 'No thread was created. Use workspaceStrategy {type:"root"} instead.', + ); + expect(launched).toHaveLength(0); + expect((yield* launchFrom("main")).at(-1)?.result).toMatchObject({ projectId }); expect(launched).toHaveLength(1); }), diff --git a/apps/server/src/scheduledTasks/ScheduledTaskService.test.ts b/apps/server/src/scheduledTasks/ScheduledTaskService.test.ts index b9a45dcfdbd6..f1426c245db6 100644 --- a/apps/server/src/scheduledTasks/ScheduledTaskService.test.ts +++ b/apps/server/src/scheduledTasks/ScheduledTaskService.test.ts @@ -3,16 +3,18 @@ import * as NodeUtil from "node:util"; import * as NodeCrypto from "@effect/platform-node/NodeCrypto"; import { assert, it } from "@effect/vitest"; -import { ScheduledTaskError, ScheduledTaskUpsertInput } from "@t3tools/contracts"; +import { RunId, ScheduledTaskError, ScheduledTaskUpsertInput, ThreadId } from "@t3tools/contracts"; import * as DateTime from "effect/DateTime"; import * as Effect from "effect/Effect"; import * as Layer from "effect/Layer"; +import * as Option from "effect/Option"; import * as Logger from "effect/Logger"; import * as SqlClient from "effect/sql/SqlClient"; import * as Deferred from "effect/Deferred"; import * as Ref from "effect/Ref"; import * as Schema from "effect/Schema"; +import * as Stream from "effect/Stream"; import * as TestClock from "effect/testing/TestClock"; import * as ThreadLaunchService from "../orchestration-v2/ThreadLaunchService.ts"; @@ -395,3 +397,65 @@ it.effect("records a run whose worktree cannot be made as failed instead of laun }).pipe(Effect.provide(ScheduledTaskService.layer.pipe(Layer.provide(layerDependencies)))); }).pipe(Effect.provide(SqlitePersistence.layerMemory)), ); + +it.effect("marks a run failed when its workspace preparation fails after launch", () => + Effect.gen(function* () { + const threadId = ThreadId.make("thread:scheduled-preparation"); + const runId = RunId.make("run:scheduled-preparation"); + const messageId = yield* Deferred.make(); + const reads = yield* Ref.make(0); + const layerDependencies = Layer.mergeAll( + NodeCrypto.layer, + Scheduler.layer, + Layer.mock(ThreadLaunchService.ThreadLaunchService)({ + checkWorktreeBase: () => Effect.void, + launch: (input) => + Deferred.succeed(messageId, input.initialMessage?.messageId ?? "").pipe( + Effect.as({ threadId, projection: {} as never, resumed: false }), + ), + }), + Layer.mock(ThreadManagementService.ThreadManagementService)({ + getThreadEventSequence: () => Effect.succeed(0), + // Preparing when the scheduler first looks, failed once the run moves on. + getThreadRecords: () => + Effect.all([Deferred.await(messageId), Ref.getAndUpdate(reads, (count) => count + 1)]).pipe( + Effect.map( + ([userMessageId, count]) => + ({ + runs: [{ id: runId, userMessageId, status: count === 0 ? "preparing" : "failed" }], + }) as never, + ), + ), + streamStoredEventsFrom: () => + Stream.make({ + event: { type: "run.updated", payload: { id: runId, status: "failed" } }, + } as never), + }), + Layer.mock(SecretRequests.SecretRequests)({}), + ); + yield* Effect.gen(function* () { + const service = yield* ScheduledTaskService.ScheduledTaskService; + const { task } = yield* service.upsert( + yield* decodeUpsertInput({ + title: "Nightly check", + prompt: "Check the build.", + enabled: true, + schedule: { type: "interval", everyMs: 60_000 }, + projectId: "project-failing-setup", + workspaceStrategy: { type: "worktree", baseRef: "main", startFromOrigin: true }, + modelSelection: { instanceId: "codex", model: "gpt-5.4" }, + runtimeMode: "full-access", + interactionMode: "default", + }), + ); + const { task: ran } = yield* service.runNow({ id: task.id }); + assert.equal(ran.lastRunStatus, "succeeded"); + const failed = yield* service.subscribeList().pipe( + Stream.map((list) => list.tasks[0]), + Stream.filter((listed) => listed?.lastRunStatus === "failed"), + Stream.runHead, + ); + assert.include(Option.getOrUndefined(failed)?.lastRunError, "Workspace preparation failed"); + }).pipe(Effect.provide(ScheduledTaskService.layer.pipe(Layer.provide(layerDependencies)))); + }).pipe(Effect.provide(SqlitePersistence.layerMemory)), +); diff --git a/apps/server/src/scheduledTasks/ScheduledTaskService.ts b/apps/server/src/scheduledTasks/ScheduledTaskService.ts index 8fd411ff2d72..fe9e481cfe54 100644 --- a/apps/server/src/scheduledTasks/ScheduledTaskService.ts +++ b/apps/server/src/scheduledTasks/ScheduledTaskService.ts @@ -674,6 +674,63 @@ export const layer = Layer.effect( ), ); + // A launch is accepted before its workspace is prepared, so a run counts as + // succeeded first. Wait for that run to leave preparation and record a + // failed preparation against this run only: last_run_at is the run's + // identity, so a later run's status is never overwritten. + const recordFailedPreparation = Effect.fn("ScheduledTaskService.recordFailedPreparation")( + function* (input: { + readonly id: ScheduledTaskId; + readonly startedAtIso: string; + readonly threadId: ThreadId; + readonly messageId: MessageId; + }) { + const afterSequence = yield* threadManagement.getThreadEventSequence(input.threadId); + const readRun = threadManagement + .getThreadRecords(input.threadId, ["runs"]) + .pipe( + Effect.map((records) => + records.runs.find((run) => run.userMessageId === input.messageId), + ), + ); + let run = yield* readRun; + if (run?.status === "preparing") { + const runId = run.id; + yield* threadManagement + .streamStoredEventsFrom({ + threadId: input.threadId, + afterSequence, + eventType: "run.updated", + }) + .pipe( + Stream.filter( + (stored) => + stored.event.type === "run.updated" && + stored.event.payload.id === runId && + stored.event.payload.status !== "preparing", + ), + Stream.runHead, + ); + run = yield* readRun; + } + if (run?.status !== "failed") return; + yield* sql` + UPDATE scheduled_tasks + SET last_run_status = 'failed', + last_run_error = ${`Workspace preparation failed. See thread ${input.threadId}.`} + WHERE task_id = ${input.id} + AND last_run_status = 'succeeded' + AND last_run_at = ${input.startedAtIso} + `; + yield* notifyChanged; + }, + Effect.catchCause((cause) => + Cause.hasInterruptsOnly(cause) + ? Effect.void + : Effect.logWarning("Could not record a failed workspace preparation", { cause }), + ), + ); + // Best-effort escape hatch: if anything fails between markRunning and // markCompleted, write a full terminal record so runDueTasks neither skips // the task forever (it filters out 'running' rows) nor re-fires it @@ -802,41 +859,45 @@ export const layer = Layer.effect( : Effect.void ).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, - }), + 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)), ), ), ) : yield* Effect.exit( - threadManagement.sendToThread({ - projectId: active.projectId, - commandId, - threadId: ThreadId.make(active.threadId), - messageId, - scheduledTaskId: active.id, - text: prompt, - attachments: [], - modelSelection: active.modelSelection, - // Scheduled prompts must not interrupt tools in the bound thread. - mode: "queue", - createdBy: active.createdBy, - creationSource: active.creationSource, - }), + threadManagement + .sendToThread({ + projectId: active.projectId, + commandId, + threadId: ThreadId.make(active.threadId), + messageId, + scheduledTaskId: active.id, + text: prompt, + attachments: [], + modelSelection: active.modelSelection, + // Scheduled prompts must not interrupt tools in the bound thread. + mode: "queue", + createdBy: active.createdBy, + creationSource: active.creationSource, + }) + .pipe(Effect.as(null)), ); const completedAt = yield* localNow; @@ -869,6 +930,15 @@ export const layer = Layer.effect( startedAtIso, }); yield* notifyChanged; + const launchedThreadId = result._tag === "Success" ? result.value : null; + if (launchedThreadId !== null) { + yield* recordFailedPreparation({ + id: task.id, + startedAtIso, + threadId: launchedThreadId, + messageId, + }).pipe(Effect.forkIn(serviceScope)); + } } return completed; }).pipe( From 2911fbdaa6fdbdabcdd8edef71b6c6d0f048e3e9 Mon Sep 17 00:00:00 2001 From: Hasan Date: Thu, 15 Oct 2026 11:08:15 +0400 Subject: [PATCH 4/5] fix(server): judge a scheduled run's preparation by its first exit The watcher read the run's current status, so a retried preparation could hide its first failure, and a provider failure after setup could be blamed on setup. It now replays the new thread's run updates and decides from the run's first status other than "preparing". Co-Authored-By: Claude Opus 5.5 --- .../ScheduledTaskService.test.ts | 95 ++++++++++++++----- .../scheduledTasks/ScheduledTaskService.ts | 48 ++++------ 2 files changed, 92 insertions(+), 51 deletions(-) diff --git a/apps/server/src/scheduledTasks/ScheduledTaskService.test.ts b/apps/server/src/scheduledTasks/ScheduledTaskService.test.ts index f1426c245db6..6d6b6c46b584 100644 --- a/apps/server/src/scheduledTasks/ScheduledTaskService.test.ts +++ b/apps/server/src/scheduledTasks/ScheduledTaskService.test.ts @@ -398,12 +398,18 @@ it.effect("records a run whose worktree cannot be made as failed instead of laun }).pipe(Effect.provide(SqlitePersistence.layerMemory)), ); -it.effect("marks a run failed when its workspace preparation fails after launch", () => +// Runs one worktree task whose launched thread reports the given run updates, +// from this run unless marked as another run's. +const runWithRunUpdates = ( + updates: Stream.Stream<{ readonly status: string; readonly otherRun?: true }>, + afterRun: ( + service: ScheduledTaskService.ScheduledTaskService["Service"], + ) => Effect.Effect, +) => Effect.gen(function* () { const threadId = ThreadId.make("thread:scheduled-preparation"); const runId = RunId.make("run:scheduled-preparation"); const messageId = yield* Deferred.make(); - const reads = yield* Ref.make(0); const layerDependencies = Layer.mergeAll( NodeCrypto.layer, Scheduler.layer, @@ -415,21 +421,28 @@ it.effect("marks a run failed when its workspace preparation fails after launch" ), }), Layer.mock(ThreadManagementService.ThreadManagementService)({ - getThreadEventSequence: () => Effect.succeed(0), - // Preparing when the scheduler first looks, failed once the run moves on. - getThreadRecords: () => - Effect.all([Deferred.await(messageId), Ref.getAndUpdate(reads, (count) => count + 1)]).pipe( - Effect.map( - ([userMessageId, count]) => - ({ - runs: [{ id: runId, userMessageId, status: count === 0 ? "preparing" : "failed" }], - }) as never, + streamStoredEventsFrom: () => + Stream.unwrap( + Deferred.await(messageId).pipe( + Effect.map((userMessageId) => + updates.pipe( + Stream.map( + ({ status, otherRun }) => + ({ + event: { + type: "run.updated", + payload: { + id: runId, + userMessageId: otherRun ? "message:other-run" : userMessageId, + status, + }, + }, + }) as never, + ), + ), + ), ), ), - streamStoredEventsFrom: () => - Stream.make({ - event: { type: "run.updated", payload: { id: runId, status: "failed" } }, - } as never), }), Layer.mock(SecretRequests.SecretRequests)({}), ); @@ -450,12 +463,50 @@ it.effect("marks a run failed when its workspace preparation fails after launch" ); const { task: ran } = yield* service.runNow({ id: task.id }); assert.equal(ran.lastRunStatus, "succeeded"); - const failed = yield* service.subscribeList().pipe( - Stream.map((list) => list.tasks[0]), - Stream.filter((listed) => listed?.lastRunStatus === "failed"), - Stream.runHead, - ); - assert.include(Option.getOrUndefined(failed)?.lastRunError, "Workspace preparation failed"); + yield* afterRun(service); }).pipe(Effect.provide(ScheduledTaskService.layer.pipe(Layer.provide(layerDependencies)))); - }).pipe(Effect.provide(SqlitePersistence.layerMemory)), + }).pipe(Effect.provide(SqlitePersistence.layerMemory)); + +it.effect("marks a run failed when its workspace preparation fails after launch", () => + // Another run moving on does not count. Preparation fails, then the user + // retries it: the run's first exit from preparation decides. + runWithRunUpdates( + Stream.make( + { status: "starting", otherRun: true } as const, + { status: "preparing" }, + { status: "failed" }, + { status: "preparing" }, + ), + (service) => + Effect.gen(function* () { + const failed = yield* service.subscribeList().pipe( + Stream.map((list) => list.tasks[0]), + Stream.filter((listed) => listed?.lastRunStatus === "failed"), + Stream.runHead, + ); + assert.include(Option.getOrUndefined(failed)?.lastRunError, "Workspace preparation failed"); + }), + ), +); + +it.effect("leaves a run succeeded when the provider fails after preparation", () => + Effect.gen(function* () { + const closed = yield* Deferred.make(); + const readFailure = yield* Deferred.make(); + // This run's later "failed" is produced only if read, so reading past + // "starting" is visible. + const updates = Stream.concat( + Stream.make({ status: "preparing" }, { status: "starting" }), + Stream.fromEffect( + Deferred.succeed(readFailure, undefined).pipe(Effect.as({ status: "failed" })), + ), + ).pipe(Stream.ensuring(Deferred.succeed(closed, undefined))); + yield* runWithRunUpdates(updates, (service) => + Effect.gen(function* () { + yield* Deferred.await(closed); + assert.isFalse(yield* Deferred.isDone(readFailure)); + assert.equal((yield* service.list()).tasks[0]?.lastRunStatus, "succeeded"); + }), + ); + }), ); diff --git a/apps/server/src/scheduledTasks/ScheduledTaskService.ts b/apps/server/src/scheduledTasks/ScheduledTaskService.ts index fe9e481cfe54..bca502390c6d 100644 --- a/apps/server/src/scheduledTasks/ScheduledTaskService.ts +++ b/apps/server/src/scheduledTasks/ScheduledTaskService.ts @@ -675,9 +675,10 @@ export const layer = Layer.effect( ); // A launch is accepted before its workspace is prepared, so a run counts as - // succeeded first. Wait for that run to leave preparation and record a - // failed preparation against this run only: last_run_at is the run's - // identity, so a later run's status is never overwritten. + // succeeded first. The first time the launched run leaves preparation says + // how preparation ended; a later retry or provider failure is not about the + // workspace. A failure is recorded against this run only: last_run_at is + // the run's identity, so a later run's status is never overwritten. const recordFailedPreparation = Effect.fn("ScheduledTaskService.recordFailedPreparation")( function* (input: { readonly id: ScheduledTaskId; @@ -685,35 +686,24 @@ export const layer = Layer.effect( readonly threadId: ThreadId; readonly messageId: MessageId; }) { - const afterSequence = yield* threadManagement.getThreadEventSequence(input.threadId); - const readRun = threadManagement - .getThreadRecords(input.threadId, ["runs"]) + // The thread is new, so replaying its run updates from the start is cheap. + const leftPreparation = yield* threadManagement + .streamStoredEventsFrom({ + threadId: input.threadId, + afterSequence: 0, + eventType: "run.updated", + }) .pipe( - Effect.map((records) => - records.runs.find((run) => run.userMessageId === input.messageId), + Stream.map((stored) => + stored.event.type === "run.updated" && + stored.event.payload.userMessageId === input.messageId + ? stored.event.payload.status + : "preparing", ), + Stream.filter((status) => status !== "preparing"), + Stream.runHead, ); - let run = yield* readRun; - if (run?.status === "preparing") { - const runId = run.id; - yield* threadManagement - .streamStoredEventsFrom({ - threadId: input.threadId, - afterSequence, - eventType: "run.updated", - }) - .pipe( - Stream.filter( - (stored) => - stored.event.type === "run.updated" && - stored.event.payload.id === runId && - stored.event.payload.status !== "preparing", - ), - Stream.runHead, - ); - run = yield* readRun; - } - if (run?.status !== "failed") return; + if (Option.getOrUndefined(leftPreparation) !== "failed") return; yield* sql` UPDATE scheduled_tasks SET last_run_status = 'failed', From e0fd539e0160ccf05c775e5f7e572167c4a3a79a Mon Sep 17 00:00:00 2001 From: Hasan Date: Thu, 15 Oct 2026 11:37:17 +0400 Subject: [PATCH 5/5] style(server): format the launch handler test Co-Authored-By: Claude Opus 5.5 --- .../src/mcp/toolkits/project/handlers.test.ts | 18 +++++++++--------- 1 file changed, 9 insertions(+), 9 deletions(-) diff --git a/apps/server/src/mcp/toolkits/project/handlers.test.ts b/apps/server/src/mcp/toolkits/project/handlers.test.ts index 732055452fb4..72add804cf13 100644 --- a/apps/server/src/mcp/toolkits/project/handlers.test.ts +++ b/apps/server/src/mcp/toolkits/project/handlers.test.ts @@ -484,15 +484,15 @@ it.effect("a launch into a new worktree creates nothing when its base ref has no }), ) : workspaceStrategy.type === "worktree" && workspaceStrategy.baseRef === "t3/renamed" - ? Effect.fail( - new ThreadLaunch.ThreadLaunchBaseRefError({ - projectId, - projectTitle: "Client target", - workspaceRoot: "/projects/client-target", - baseRef: workspaceStrategy.baseRef, - }), - ) - : Effect.void, + ? Effect.fail( + new ThreadLaunch.ThreadLaunchBaseRefError({ + projectId, + projectTitle: "Client target", + workspaceRoot: "/projects/client-target", + baseRef: workspaceStrategy.baseRef, + }), + ) + : Effect.void, }); const toolkit = yield* ProjectToolkit.pipe( Effect.provide(