diff --git a/apps/server/src/scheduledTasks/ScheduledTaskService.test.ts b/apps/server/src/scheduledTasks/ScheduledTaskService.test.ts index ee4385ff2e0af..1335e369484a5 100644 --- a/apps/server/src/scheduledTasks/ScheduledTaskService.test.ts +++ b/apps/server/src/scheduledTasks/ScheduledTaskService.test.ts @@ -4,6 +4,7 @@ import * as NodeUtil from "node:util"; import * as NodeCrypto from "@effect/platform-node/NodeCrypto"; import { assert, it, vi } from "@effect/vitest"; import { + CommandId, ProjectId, ProviderInstanceId, ScheduledTaskError, @@ -29,6 +30,7 @@ import * as TestClock from "effect/testing/TestClock"; import * as ThreadLaunchService from "../orchestration-v2/ThreadLaunchService.ts"; import * as ThreadManagementService from "../orchestration-v2/ThreadManagementService.ts"; +import { OrchestratorDispatchError } from "../orchestration-v2/Orchestrator.ts"; import * as SecretRequests from "../secrets/SecretRequests.ts"; import * as ManagedProjectFolders from "../project/ManagedProjectFolders.ts"; import * as SqlitePersistence from "../persistence/Sqlite.ts"; @@ -663,6 +665,67 @@ it.effect("a fresh-session task launches one thread, then starts each later run }).pipe(Effect.provide(SqlitePersistence.layerMemory)), ); +it.effect("a refused dispatch records its refusal, and opaque causes stay generic", () => + Effect.gen(function* () { + const refusal = "Thread thread:refused was stopped."; + const refused = (cause: unknown) => + new OrchestratorDispatchError({ + commandId: CommandId.make("command:refused"), + commandType: "message.dispatch", + cause, + }); + const generic = refused(undefined).message; + const cases = [ + [refusal, refusal], + [" ", generic], + [new Error("private-nested-secret"), generic], + ] as const; + const failure = yield* Ref.make(undefined); + const layer = ScheduledTaskService.layer.pipe( + Layer.provide( + Layer.mergeAll( + ManagedPolicy.layerTest(), + Layer.mock(ThreadLaunchService.ThreadLaunchService)({}), + Layer.mock(ThreadManagementService.ThreadManagementService)({ + getThreadShell: () => Effect.succeed(threadShell(ProjectId.make("project:test"))), + sendToThread: () => + Ref.get(failure).pipe(Effect.flatMap((cause) => Effect.fail(refused(cause)))), + }), + Layer.mock(SecretRequests.SecretRequests)({}), + Layer.mock(ManagedProjectFolders.ManagedProjectFolders)({ + namedProjectsRoot: "/projects", + isScratchProject: () => Effect.succeed(false), + }), + NodeCrypto.layer, + Layer.succeed(Scheduler.Scheduler, { register: () => Effect.void }), + ), + ), + ); + yield* Effect.gen(function* () { + const service = yield* ScheduledTaskService.ScheduledTaskService; + const { task } = yield* service.upsert({ + id: ScheduledTaskId.make("task:refused"), + title: "Refused dispatch", + prompt: "Check progress", + enabled: true, + schedule: { type: "interval", everyMs: 60_000 }, + projectId: ProjectId.make("project:test"), + threadId: ThreadId.make("thread:refused"), + workspaceStrategy: { type: "root" }, + modelSelection: { instanceId: ProviderInstanceId.make("codex"), model: "gpt-5.4" }, + runtimeMode: "full-access", + interactionMode: "default", + }); + for (const [cause, expected] of cases) { + yield* Ref.set(failure, cause); + const result = yield* service.runNow({ id: task.id }); + assert.equal(result.task.lastRunStatus, "failed"); + assert.equal(result.task.lastRunError, expected); + } + }).pipe(Effect.provide(layer)); + }).pipe(Effect.provide(SqlitePersistence.layerMemory)), +); + it.effect("a first run whose thread was deleted before it bound removes the task", () => Effect.gen(function* () { const layer = ScheduledTaskService.layer.pipe( diff --git a/apps/server/src/scheduledTasks/ScheduledTaskService.ts b/apps/server/src/scheduledTasks/ScheduledTaskService.ts index 05fdbe8b08992..9ed134be2da5a 100644 --- a/apps/server/src/scheduledTasks/ScheduledTaskService.ts +++ b/apps/server/src/scheduledTasks/ScheduledTaskService.ts @@ -53,6 +53,8 @@ import * as ThreadLaunchService from "../orchestration-v2/ThreadLaunchService.ts import * as ManagedProjectFolders from "../project/ManagedProjectFolders.ts"; import * as Metrics from "../observability/Metrics.ts"; import * as ThreadManagementService from "../orchestration-v2/ThreadManagementService.ts"; +import { OrchestratorDispatchError } from "../orchestration-v2/Orchestrator.ts"; +import { userFacingDispatchErrorMessage } from "../orchestration-v2/UserFacingErrors.ts"; import * as SecretRequests from "../secrets/SecretRequests.ts"; import * as ManagedPolicy from "../policy/ManagedPolicy.ts"; import * as DueHint from "../scheduling/DueHint.ts"; @@ -79,6 +81,7 @@ import { constantTimeEquals, verifyWebhookSignature } from "./webhookVerificatio export const WEBHOOK_ROUTE_PREFIX = "/api/hooks"; const isScheduledTaskError = Schema.is(ScheduledTaskError); +const isOrchestratorDispatchError = Schema.is(OrchestratorDispatchError); const PROMPT_TOO_LONG = "The filled-in prompt is too long."; const WEBHOOK_DELIVERY_RETENTION = 50; const WEBHOOK_DEDUPE_DELIVERY_RETENTION = 20; @@ -364,6 +367,14 @@ function errorMessage(error: unknown): string { return failure instanceof Error ? failure.message : String(failure); } +function dispatchErrorMessage(cause: Cause.Cause): string { + const failure = Cause.squash(cause); + // Direct orchestration refusals are user-facing; nested defects can contain private data. + return isOrchestratorDispatchError(failure) && typeof failure.cause === "string" + ? (userFacingDispatchErrorMessage(failure.cause) ?? errorMessage(failure)) + : errorMessage(failure); +} + function webhookPath(taskId: string, token: string): string { return `${WEBHOOK_ROUTE_PREFIX}/${encodeURIComponent(taskId)}/${token}`; } @@ -1098,7 +1109,7 @@ export const layer = Layer.effect( 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); + const lastRunError = runSucceeded ? null : dispatchErrorMessage(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